You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2020/12/15 04:56:38 UTC
[GitHub] [iceberg] zhangjun0x01 opened a new pull request #1936: Flink : add parallelism optimize for IcebergTableSource
zhangjun0x01 opened a new pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936
When using flink to query the iceberg table, the parallelism is the default parallelism of flink, but the number of datafiles on iceberg table is different. The user do not know how much parallelism should be used, and setting a too large parallelism will cause resource waste, setting the parallelism too small will cause the query to be slow, so we can add parallelism infer.
The function is enabled by default. the parallelism is equal to the number of data files. Of course, the user can manually turn off the infer function. In order to prevent too many datafiles from causing excessive parallelism, we also set a max infer parallelism. When the infer parallelism exceeds the setting, use the max parallelism.
In addition, we also need to compare with the limit in the `select` query statement to get a more appropriate parallelism in the case of limit pushdown, for example we have a sql `select * from table limit 1`, and finally we infer the parallelism is 10, but we only one parallel is needed , besause we only need a 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r561461670
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
I think these are two different concepts.
In flink, whether using batch mode or streaming mode, we can read batch data. Flink treats batch jobs as bounded streaming jobs, so there should be no problem whether it is using batch mode or streaming mode to read batch data. In addition, flink will use `StreamExecutionEnvironment (DataStream)` to do batch tasks and stream tasks uniformly ([the doc link](https://flink.apache.org/news/2020/12/10/release-1.12.0.html#batch-execution-mode-in-the-datastream-api)) . The batch mode may expire, so I think we should also use `StreamExecutionEnvironment (DataStream)` for batch tasks as much as possible.
When we use `StreamExecutionEnvironment` , in `FlinkSource.Builder#build` method,
the `if` and `else` block in this method are both streaming jobs, `if` block is a bounded streaming jobs, maybe we can rename `ScanContext#isStreaming` field to `ScanContext#isStreamingRead`, which will be easier to understand. `If` code block is bounded stream job (batch), `else` code block to do long-running stream job.
```
if (!context.isStreaming()) {
int parallelism = inferParallelism(format, context);
return env.createInput(format, typeInfo).setParallelism(parallelism);
} else {
StreamingMonitorFunction function = new StreamingMonitorFunction(tableLoader, context);
String monitorFunctionName = String.format("Iceberg table (%s) monitor", table);
String readerOperatorName = String.format("Iceberg table (%s) reader", table);
return env.addSource(function, monitorFunctionName)
.transform(readerOperatorName, typeInfo, StreamingReaderOperator.factory(format));
}
```
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
> In my mind, Providing unit tests to check whether the `inferParallelism()` is returning the expected parallelism value is enough for this changes. Seems like The ITCase is validating the behavior of DataStreamSource#setParallelism , we could think it's always correct because it's a basic API in flink.
I think it’s better not to use the `inferParallelism` method to get the parallelism to do assertion, because the `inferParallelism` method is private and is an internal method of iceberg. Just as you [commented](https://github.com/apache/iceberg/pull/1936#discussion_r554983424) that it is best not to use the internal code of flink, I think we should try to use public APIs to get information.
The current `TestFlinkTableSource` class uses batch mode for unit test. In order not to modify too much code, we can move the `testInferedParallelism` method to other test classes, such as `TestFlinkScan.java`.
So I think we can use `DataStream.getTransformation().getParallelism();` to get the parallelism of the flink operator. This method is public api of flink. Even if flink is upgraded in the future, it should not be modified. What do you think?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -70,6 +73,7 @@ public static Builder forRowData() {
private Table table;
private TableLoader tableLoader;
private TableSchema projectedSchema;
+ private ReadableConfig flinkConf;
Review comment:
When we construct the `IcebergTableSource`, we use the `TableSourceFactory.Context#getConfiguration` method to get the configuration. This method returns a ReadableConfig, so we use ReadableConfig instead of Configuration. In addition, Configuration is the implementation class of the ReadableConfig interface, so I think ReadableConfig should not has a problem
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
yes,it should be a streaming source, like kafka. If necessary, we can open a separate PR to discuss this.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
I update the pr,move the `testInferedParallelism` method to TestFlinkScanSql,use `FlinkSource.Builder#inferParallelism` method to do the assertion
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r554950693
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkTableOptions.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.iceberg.flink;
+
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+public class FlinkTableOptions {
+
+ private FlinkTableOptions() {
+ }
+
+ public static final ConfigOption<Boolean> TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM =
+ ConfigOptions.key("table.exec.iceberg.infer-source-parallelism").defaultValue(true)
Review comment:
The `defaultValue(true)` says deprecated now. Let's change it to:
```java
public static final ConfigOption<Boolean> TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM =
ConfigOptions.key("table.exec.iceberg.infer-source-parallelism").booleanType().defaultValue(true)
.withDescription("If is false, parallelism of source are set by config.\n" +
"If is true, source parallelism is inferred according to splits number.\n");
```
The similar thing in `TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX`.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -195,7 +205,29 @@ public FlinkInputFormat buildFormat() {
Preconditions.checkNotNull(env, "StreamExecutionEnvironment should not be null");
FlinkInputFormat format = buildFormat();
if (isBounded(context)) {
- return env.createInput(format, rowTypeInfo);
+ int parallelism = flinkConf.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
+ if (flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
+ int max = flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
+ if (max < 1) {
+ throw new IllegalConfigurationException(
+ FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX.key() + " cannot be less than 1");
+ }
+
+ int splitNum = 0;
+ try {
+ FlinkInputSplit[] splits = format.createInputSplits(0);
+ splitNum = splits.length;
+ } catch (IOException e) {
+ throw new RuntimeException("get input split error.", e);
Review comment:
Nit: use `UncheckedIOException` here.
```java
throw new UncheckedIOException("Failed to create iceberg input splits for table: " + table, e);
```
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -195,7 +205,29 @@ public FlinkInputFormat buildFormat() {
Preconditions.checkNotNull(env, "StreamExecutionEnvironment should not be null");
FlinkInputFormat format = buildFormat();
if (isBounded(context)) {
- return env.createInput(format, rowTypeInfo);
+ int parallelism = flinkConf.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
Review comment:
How about moving those lines into a separate method ?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -195,7 +205,29 @@ public FlinkInputFormat buildFormat() {
Preconditions.checkNotNull(env, "StreamExecutionEnvironment should not be null");
FlinkInputFormat format = buildFormat();
if (isBounded(context)) {
- return env.createInput(format, rowTypeInfo);
+ int parallelism = flinkConf.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
+ if (flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
+ int max = flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
+ if (max < 1) {
+ throw new IllegalConfigurationException(
+ FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX.key() + " cannot be less than 1");
+ }
+
+ int splitNum = 0;
+ try {
+ FlinkInputSplit[] splits = format.createInputSplits(0);
+ splitNum = splits.length;
+ } catch (IOException e) {
+ throw new RuntimeException("get input split error.", e);
+ }
+
+ parallelism = Math.min(splitNum, max);
+ }
+
+ parallelism = limit > 0 ? Math.min(parallelism, (int) limit) : parallelism;
Review comment:
It may be overflow when casting the long `limit` to integer ? I'd like to use `(int) Math.min(parallelism, limit)`.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -103,4 +124,45 @@ public void testLimitPushDown() {
Assert.assertEquals("should have 1 record", 1, mixedResult.size());
Assert.assertArrayEquals("Should produce the expected records", mixedResult.get(0), new Object[] {1, "a"});
}
+
+ @Test
+ public void testParallelismOptimize() {
+ sql("INSERT INTO %s VALUES (1,'hello')", TABLE_NAME);
+ sql("INSERT INTO %s VALUES (2,'iceberg')", TABLE_NAME);
+
+ TableEnvironment tenv = getTableEnv();
+
+ // empty table ,parallelism at least 1
+ Table tableEmpty = tenv.sqlQuery(String.format("SELECT * FROM %s", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(1, tableEmpty, tenv);
+
+ // make sure to generate 2 CombinedScanTasks
+ org.apache.iceberg.Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
+ Stream<FileScanTask> stream = StreamSupport.stream(table.newScan().planFiles().spliterator(), false);
+ Optional<FileScanTask> fileScanTaskOptional = stream.max(Comparator.comparing(FileScanTask::length));
+ Assert.assertTrue(fileScanTaskOptional.isPresent());
+ long maxFileLen = fileScanTaskOptional.get().length();
+ sql("ALTER TABLE %s SET ('read.split.open-file-cost'='1', 'read.split.target-size'='%s')", TABLE_NAME, maxFileLen);
+
+ // 2 splits ,the parallelism is 2
+ Table tableSelect = tenv.sqlQuery(String.format("SELECT * FROM %s", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(2, tableSelect, tenv);
+
+ // 2 splits and limit is 1 ,the parallelism is 1
+ Table tableLimit = tenv.sqlQuery(String.format("SELECT * FROM %s LIMIT 1", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(1, tableLimit, tenv);
+ }
+
+ private void testParallelismSettingTranslateAndAssert(int expected, Table table, TableEnvironment tEnv) {
Review comment:
Is there another way to assert the parallelism as expected value ? Here we're using flink's planner to get the `ExecNode` , I'm concerning that we're using flink's `Internal` codes which would be a big trouble when upgrading the flink version. Pls see this PR https://github.com/apache/iceberg/pull/1956
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -195,7 +205,29 @@ public FlinkInputFormat buildFormat() {
Preconditions.checkNotNull(env, "StreamExecutionEnvironment should not be null");
FlinkInputFormat format = buildFormat();
if (isBounded(context)) {
- return env.createInput(format, rowTypeInfo);
+ int parallelism = flinkConf.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
+ if (flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
+ int max = flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
+ if (max < 1) {
+ throw new IllegalConfigurationException(
+ FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX.key() + " cannot be less than 1");
+ }
Review comment:
nit: Preconditions.checkState ?
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -103,4 +124,45 @@ public void testLimitPushDown() {
Assert.assertEquals("should have 1 record", 1, mixedResult.size());
Assert.assertArrayEquals("Should produce the expected records", mixedResult.get(0), new Object[] {1, "a"});
}
+
+ @Test
+ public void testParallelismOptimize() {
Review comment:
nit: `testParallelismOptimize` -> `testInferedParallelism`
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -103,4 +124,45 @@ public void testLimitPushDown() {
Assert.assertEquals("should have 1 record", 1, mixedResult.size());
Assert.assertArrayEquals("Should produce the expected records", mixedResult.get(0), new Object[] {1, "a"});
}
+
+ @Test
+ public void testParallelismOptimize() {
+ sql("INSERT INTO %s VALUES (1,'hello')", TABLE_NAME);
+ sql("INSERT INTO %s VALUES (2,'iceberg')", TABLE_NAME);
+
+ TableEnvironment tenv = getTableEnv();
+
+ // empty table ,parallelism at least 1
+ Table tableEmpty = tenv.sqlQuery(String.format("SELECT * FROM %s", TABLE_NAME));
Review comment:
nit: how about introducing a small method:
```java
private Table sqlQuery(String sql, Object... args) {
return getTableEnv().sqlQuery(String.format(sql, args));
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r561513015
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
yes,it should be a streaming source, like kafka. If necessary, we can open a separate PR to discuss 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r562450373
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -70,6 +73,7 @@ public static Builder forRowData() {
private Table table;
private TableLoader tableLoader;
private TableSchema projectedSchema;
+ private ReadableConfig readableConfig;
Review comment:
yes,I add the `new Configuration()` for default.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r560857549
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
I found that in this test method,I use the flink streaming mode,but it still enter the batch mode ([here](https://github.com/apache/iceberg/blob/master/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java#L200)), I check the code,found that `FlinkSource.Builder#build` mthod judge streaming mode or batch mode by the conf of ScanContext instead of flink conf. will this confuse users?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r562450034
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -106,6 +109,51 @@ public void testResiduals() throws Exception {
assertRecords(runWithFilter(filter, "where dt='2020-03-20' and id=123"), expectedRecords, SCHEMA);
}
+ @Test
+ public void testInferedParallelism() throws IOException {
+ Table table = catalog.createTable(TableIdentifier.of("default", "t"), SCHEMA, SPEC);
+
+ TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
+ FlinkInputFormat flinkInputFormat = FlinkSource.forRowData().tableLoader(tableLoader).table(table).buildFormat();
+ ScanContext scanContext = ScanContext.builder().build();
+
+ // Empty table ,parallelism at least 1
+ int parallelism = FlinkSource.forRowData()
+ .flinkConf(new Configuration())
+ .inferParallelism(flinkInputFormat, scanContext);
+ Assert.assertEquals("Should produce the expected parallelism.", 1, parallelism);
+
+ List<Record> writeRecords = RandomGenericData.generate(SCHEMA, 2, 0L);
+ writeRecords.get(0).set(1, 123L);
+ writeRecords.get(0).set(2, "2020-03-20");
+ writeRecords.get(1).set(1, 456L);
+ writeRecords.get(1).set(2, "2020-03-20");
+
+ GenericAppenderHelper helper = new GenericAppenderHelper(table, fileFormat, TEMPORARY_FOLDER);
+
+ DataFile dataFile1 = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0), writeRecords);
+ DataFile dataFile2 = helper.writeFile(TestHelpers.Row.of("2020-03-21", 0),
+ RandomGenericData.generate(SCHEMA, 2, 0L));
Review comment:
At first I copy the code from `TestFlinkScanSql#testResiduals` method to gererate 2 datafiles.
I think there should be no problem about the partition. `writeRecords` will write to the partition `2020-03-20`, and randomly generate two records into the partition `2020-03-21`.
But for simplicity, I modified the code to randomly generate two records for each partition.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r561506300
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
The `isStreaming` indicate whether the flink source is a streaming source (In our mind) , not say it's a streaming job or batch job. The hive table source also has the similar configure key :
```java
public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
key("streaming-source.enable")
.booleanType()
.defaultValue(false)
.withDescription(
"Enable streaming source or not.\n"
+ " NOTES: Please make sure that each partition/file should be written"
+ " atomically, otherwise the reader may get incomplete data.");
```
If we think this iceberg configure key is not very clear, I think we could propose another separate PR to align with hive configure key. Let's focus on this parallelism issue here, what do you think ?
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
> Just as you commented that it is best not to use the internal code of flink, I think we should try to use public APIs to get information.
The comment that saying we'd better not use flink's Internal API because that would introduce extra upgrade complexity (new flink version may breaks those internal API so we iceberg have to adjust the codes, finally maintaining different versions of flink will bring us a lot of burden).
Writing iceberg unit tests based on our iceberg's non-public ( we usually use package-access ) method is OK because there's no extra burden .
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -70,6 +73,7 @@ public static Builder forRowData() {
private Table table;
private TableLoader tableLoader;
private TableSchema projectedSchema;
+ private ReadableConfig flinkConf;
Review comment:
Got your point, I'd prefer to use flink's `Configuration` because it will be exposed to flink developers as an API in `FlinkSource`, using the unified `Configuration` will be more straightforward for them. But as you said, the TableSourceFactory#Context is exposing the `ReadableConfig`, I also did not find a correct way to convert `ReadableConfig` to `Configuration`. OK, I think we could use `ReadableConfig` here, thanks.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -106,6 +109,51 @@ public void testResiduals() throws Exception {
assertRecords(runWithFilter(filter, "where dt='2020-03-20' and id=123"), expectedRecords, SCHEMA);
}
+ @Test
+ public void testInferedParallelism() throws IOException {
+ Table table = catalog.createTable(TableIdentifier.of("default", "t"), SCHEMA, SPEC);
+
+ TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
+ FlinkInputFormat flinkInputFormat = FlinkSource.forRowData().tableLoader(tableLoader).table(table).buildFormat();
+ ScanContext scanContext = ScanContext.builder().build();
+
+ // Empty table ,parallelism at least 1
Review comment:
Nit: `inter parallelism should be at least 1.`
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -208,6 +218,33 @@ public FlinkInputFormat buildFormat() {
.transform(readerOperatorName, typeInfo, StreamingReaderOperator.factory(format));
}
}
+
+ int inferParallelism(FlinkInputFormat format, ScanContext context) {
+ int parallelism = readableConfig.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
+ if (readableConfig.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
+ int maxInterParallelism = readableConfig.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
Review comment:
Nit: `maxInterParallelism` -> `maxInferParallelism`, seems like it's a typo ?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -79,13 +84,20 @@ public boolean isBounded() {
@Override
public TableSource<RowData> projectFields(int[] fields) {
- return new IcebergTableSource(loader, schema, properties, fields, isLimitPushDown, limit, filters);
+ return new IcebergTableSource(loader, schema, properties, fields, isLimitPushDown, limit, filters, readableConfig);
}
@Override
public DataStream<RowData> getDataStream(StreamExecutionEnvironment execEnv) {
- return FlinkSource.forRowData().env(execEnv).tableLoader(loader).project(getProjectedSchema()).limit(limit)
- .filters(filters).properties(properties).build();
+ return FlinkSource.forRowData()
Review comment:
Nit: maybe we'd better also align the orders as above commented.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -70,6 +73,7 @@ public static Builder forRowData() {
private Table table;
private TableLoader tableLoader;
private TableSchema projectedSchema;
+ private ReadableConfig readableConfig;
Review comment:
Should we provide a `new Configuration()` for this variable ? Otherwise, it will just throw NPE if people forget to provide a `flinkConf` in FlinkSource#Builder because we don't check the nullable in `interParallelism`.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -106,6 +109,51 @@ public void testResiduals() throws Exception {
assertRecords(runWithFilter(filter, "where dt='2020-03-20' and id=123"), expectedRecords, SCHEMA);
}
+ @Test
+ public void testInferedParallelism() throws IOException {
+ Table table = catalog.createTable(TableIdentifier.of("default", "t"), SCHEMA, SPEC);
+
+ TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
+ FlinkInputFormat flinkInputFormat = FlinkSource.forRowData().tableLoader(tableLoader).table(table).buildFormat();
+ ScanContext scanContext = ScanContext.builder().build();
+
+ // Empty table ,parallelism at least 1
+ int parallelism = FlinkSource.forRowData()
+ .flinkConf(new Configuration())
+ .inferParallelism(flinkInputFormat, scanContext);
+ Assert.assertEquals("Should produce the expected parallelism.", 1, parallelism);
+
+ List<Record> writeRecords = RandomGenericData.generate(SCHEMA, 2, 0L);
+ writeRecords.get(0).set(1, 123L);
+ writeRecords.get(0).set(2, "2020-03-20");
+ writeRecords.get(1).set(1, 456L);
+ writeRecords.get(1).set(2, "2020-03-20");
+
+ GenericAppenderHelper helper = new GenericAppenderHelper(table, fileFormat, TEMPORARY_FOLDER);
+
+ DataFile dataFile1 = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0), writeRecords);
+ DataFile dataFile2 = helper.writeFile(TestHelpers.Row.of("2020-03-21", 0),
+ RandomGenericData.generate(SCHEMA, 2, 0L));
+ helper.appendToTable(dataFile1, dataFile2);
+
+ // Make sure to generate 2 CombinedScanTasks
+ long maxFileLen = Math.max(dataFile1.fileSizeInBytes(), dataFile2.fileSizeInBytes());
+ executeSQL(String
+ .format("ALTER TABLE t SET ('read.split.open-file-cost'='1', 'read.split.target-size'='%s')", maxFileLen));
+
+ // 2 splits ,the parallelism is 2
+ parallelism = FlinkSource.forRowData()
Review comment:
I think there're other test cases that we don't cover, it's good to cover those tests.
1. table.exec.iceberg.infer-source-parallelism=false;
2. table.exec.iceberg.infer-source-parallelism.max <= numberOfSplits;
3. table.exec.iceberg.infer-source-parallelism.max > numberOfSplits;
4. table.exec.iceberg.infer-source-parallelism.max > limit;
5. table.exec.iceberg.infer-source-parallelism.max <= limit;
Divide those cases into small method if necessary.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -51,25 +52,29 @@
private final TableLoader loader;
private final TableSchema schema;
private final Map<String, String> properties;
+ private final ReadableConfig readableConfig;
Review comment:
Nit: Let's move this line to line60, so that the assignment order of `IcebergTableSource` constructor could align with these definitions.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -106,6 +109,51 @@ public void testResiduals() throws Exception {
assertRecords(runWithFilter(filter, "where dt='2020-03-20' and id=123"), expectedRecords, SCHEMA);
}
+ @Test
+ public void testInferedParallelism() throws IOException {
+ Table table = catalog.createTable(TableIdentifier.of("default", "t"), SCHEMA, SPEC);
+
+ TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
+ FlinkInputFormat flinkInputFormat = FlinkSource.forRowData().tableLoader(tableLoader).table(table).buildFormat();
+ ScanContext scanContext = ScanContext.builder().build();
+
+ // Empty table ,parallelism at least 1
+ int parallelism = FlinkSource.forRowData()
+ .flinkConf(new Configuration())
+ .inferParallelism(flinkInputFormat, scanContext);
+ Assert.assertEquals("Should produce the expected parallelism.", 1, parallelism);
+
+ List<Record> writeRecords = RandomGenericData.generate(SCHEMA, 2, 0L);
+ writeRecords.get(0).set(1, 123L);
+ writeRecords.get(0).set(2, "2020-03-20");
+ writeRecords.get(1).set(1, 456L);
+ writeRecords.get(1).set(2, "2020-03-20");
+
+ GenericAppenderHelper helper = new GenericAppenderHelper(table, fileFormat, TEMPORARY_FOLDER);
+
+ DataFile dataFile1 = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0), writeRecords);
+ DataFile dataFile2 = helper.writeFile(TestHelpers.Row.of("2020-03-21", 0),
+ RandomGenericData.generate(SCHEMA, 2, 0L));
Review comment:
Those random generated records will be located in partition `2020-03-21` ? I guess it's not.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -137,7 +136,10 @@ public void testFilterPushDownEqual() {
Assert.assertEquals("Should have 1 record", 1, result.size());
Assert.assertArrayEquals("Should produce the expected record", expectRecord, result.get(0));
- Assert.assertEquals("Should create only one scan", 1, scanEventCount);
+ // Because we add infer parallelism, all data files will be scanned first.
+ // Flink will call FlinkInputFormat#createInputSplits method to scan the data files,
+ // plus the operation to get the execution plan, so there are three scan event.
+ Assert.assertEquals("Should create 3 scans", 3, scanEventCount);
Review comment:
We can disable the `table.exec.iceberg.infer-source-parallelism` for all the batch tests by default, then we don't have to change all cases from this file. Actually, we have wrote many unit tests which depends on the `parallelism`, for example this PR https://github.com/apache/iceberg/pull/2064. Using the inter-parallelism for batch unit tests will introduce extra complexity and instability, so I recommend to disable the infer parallelism in our batch unit tests by default:
```diff
diff --git a/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java b/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
index 5b8e58cf..ab3d56ea 100644
--- a/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
+++ b/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
@@ -62,10 +62,17 @@ public abstract class FlinkTestBase extends AbstractTestBase {
if (tEnv == null) {
synchronized (this) {
if (tEnv == null) {
- this.tEnv = TableEnvironment.create(EnvironmentSettings
+ EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.useBlinkPlanner()
- .inBatchMode().build());
+ .inBatchMode()
+ .build();
+
+ TableEnvironment env = TableEnvironment.create(settings);
+ env.getConfig().getConfiguration()
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+
+ tEnv = env;
}
}
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r561491288
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
> In my mind, Providing unit tests to check whether the `inferParallelism()` is returning the expected parallelism value is enough for this changes. Seems like The ITCase is validating the behavior of DataStreamSource#setParallelism , we could think it's always correct because it's a basic API in flink.
I think it’s better not to use the `inferParallelism` method to get the parallelism to do assertion, because the `inferParallelism` method is private and is an internal method of iceberg. Just as you [commented](https://github.com/apache/iceberg/pull/1936#discussion_r554983424) that it is best not to use the internal code of flink, I think we should try to use public APIs to get information.
The current `TestFlinkTableSource` class uses batch mode for unit test. In order not to modify too much code, we can move the `testInferedParallelism` method to other test classes, such as `TestFlinkScan.java`.
So I think we can use `DataStream.getTransformation().getParallelism();` to get the parallelism of the flink operator. This method is public api of flink. Even if flink is upgraded in the future, it should not be modified. What do you think?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx merged pull request #1936: Flink: Support inferring parallelism for batch read.
Posted by GitBox <gi...@apache.org>.
openinx merged pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r560983546
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
Okay, that's a great point. I think it will confuse users, the correct way is : Set the ScanContext's properties firstly (use the following `fromFlinkConf`) if someone provides a flink configuration, that is similar to the ScanContext#fromProperties:
```diff
diff --git a/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java b/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
index 2896efb3..c56e3311 100644
--- a/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
+++ b/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
@@ -292,10 +292,7 @@ class ScanContext implements Serializable {
return this;
}
- Builder fromProperties(Map<String, String> properties) {
- Configuration config = new Configuration();
- properties.forEach(config::setString);
-
+ Builder fromFlinkConf(Configuration config) {
return this.useSnapshotId(config.get(SNAPSHOT_ID))
.caseSensitive(config.get(CASE_SENSITIVE))
.asOfTimestamp(config.get(AS_OF_TIMESTAMP))
@@ -305,7 +302,14 @@ class ScanContext implements Serializable {
.splitLookback(config.get(SPLIT_LOOKBACK))
.splitOpenFileCost(config.get(SPLIT_FILE_OPEN_COST))
.streaming(config.get(STREAMING))
- .monitorInterval(config.get(MONITOR_INTERVAL))
+ .monitorInterval(config.get(MONITOR_INTERVAL));
+ }
+
+ Builder fromProperties(Map<String, String> properties) {
+ Configuration config = new Configuration();
+ properties.forEach(config::setString);
+
+ return fromFlinkConf(config)
.nameMapping(properties.get(DEFAULT_NAME_MAPPING));
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r560714880
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -79,13 +84,13 @@ public boolean isBounded() {
@Override
public TableSource<RowData> projectFields(int[] fields) {
- return new IcebergTableSource(loader, schema, properties, fields, isLimitPushDown, limit, filters);
+ return new IcebergTableSource(loader, schema, properties, fields, isLimitPushDown, limit, filters, readableConfig);
}
@Override
public DataStream<RowData> getDataStream(StreamExecutionEnvironment execEnv) {
return FlinkSource.forRowData().env(execEnv).tableLoader(loader).project(getProjectedSchema()).limit(limit)
- .filters(filters).properties(properties).build();
+ .filters(filters).flinkConf(readableConfig).properties(properties).build();
Review comment:
Nit: I'd like to change this builder chain like the following ( That's more easy to read the change):
```java
@Override
public DataStream<RowData> getDataStream(StreamExecutionEnvironment execEnv) {
return FlinkSource.forRowData()
.env(execEnv)
.tableLoader(loader)
.project(getProjectedSchema())
.limit(limit)
.filters(filters)
.flinkConf(readableConfig)
.properties(properties)
.build();
}
```
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -208,6 +217,30 @@ public FlinkInputFormat buildFormat() {
.transform(readerOperatorName, typeInfo, StreamingReaderOperator.factory(format));
}
}
+
+ private DataStream<RowData> createInputDataStream(FlinkInputFormat format, ScanContext context,
+ TypeInformation<RowData> typeInfo) {
+ int parallelism = flinkConf.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
+ if (flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
+ int max = flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
+ Preconditions.checkState(max >= 1,
+ FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX.key() + " cannot be less than 1");
+ int splitNum = 0;
+ try {
+ FlinkInputSplit[] splits = format.createInputSplits(0);
+ splitNum = splits.length;
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to create iceberg input splits for table: " + table, e);
+ }
+
+ parallelism = Math.min(splitNum, max);
+ }
+
+ int limitInt = context.limit() > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) context.limit();
+ parallelism = limitInt > 0 ? Math.min(parallelism, limitInt) : parallelism;
Review comment:
Nit: I'd like to make this code more readable:
```java
if (context.limit() > 0) {
int limit = context.limit() >= Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) context.limit();
parallelism = Math.min(parallelism, limit);
}
// parallelism must be positive.
parallelism = Math.max(1, parallelism);
```
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
In this way, we don't have to change so many codes in this class. Maybe we could just add unit tests in TestFlinkScan.java
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
Shouldn't the inferParallelism only affect the batch job (See FlinkSource#Builder#build)? So there's no reason that providing unit test in streaming mode ?
In my mind, Providing unit tests to check whether the `inferParallelism()` is returning the expected parallelism value is enough for this changes. Seems like The ITCase is validating the behavior of DataStreamSource#setParallelism , we could think it's always correct because it's a basic API in flink.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -70,6 +73,7 @@ public static Builder forRowData() {
private Table table;
private TableLoader tableLoader;
private TableSchema projectedSchema;
+ private ReadableConfig flinkConf;
Review comment:
For those users that write flink batch jobs in Java API , they will always pass a flink's `Configuration`, right ? So how about defining this as `org.apache.flink.configuration.Configuraiton` ?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -197,7 +206,7 @@ public FlinkInputFormat buildFormat() {
TypeInformation<RowData> typeInfo = RowDataTypeInfo.of(FlinkSchemaUtil.convert(context.project()));
if (!context.isStreaming()) {
- return env.createInput(format, typeInfo);
+ return createInputDataStream(format, context, typeInfo);
Review comment:
In this [comment](https://github.com/apache/iceberg/pull/1936#discussion_r554954082), I think I did not describe the things clearly. I mean we could move the `inferParallelism` into a separate method, don't have to contains the DataStream constructing or chaining methods.
```java
private int inferParallelism(FlinkInputFormat format, ScanContext context) {
// ....
}
```
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkTableOptions.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.iceberg.flink;
+
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+public class FlinkTableOptions {
+
+ private FlinkTableOptions() {
+ }
+
+ public static final ConfigOption<Boolean> TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM =
+ ConfigOptions.key("table.exec.iceberg.infer-source-parallelism").booleanType().defaultValue(true)
Review comment:
Nit: it's more clear to make each option definition into a separate line:
```java
ConfigOptions.key("table.exec.iceberg.infer-source-parallelism")
.booleanType()
.defaultValue(true)
.withDescription("If is false, parallelism of source are set by config.\n" +
"If is true, source parallelism is inferred according to splits number.\n");
```
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkTableOptions.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.iceberg.flink;
+
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+public class FlinkTableOptions {
+
+ private FlinkTableOptions() {
+ }
+
+ public static final ConfigOption<Boolean> TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM =
+ ConfigOptions.key("table.exec.iceberg.infer-source-parallelism").booleanType().defaultValue(true)
+ .withDescription("If is false, parallelism of source are set by config.\n" +
+ "If is true, source parallelism is inferred according to splits number.\n");
+
+ public static final ConfigOption<Integer> TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX =
+ ConfigOptions.key("table.exec.iceberg.infer-source-parallelism.max").intType().defaultValue(100)
Review comment:
ditto
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -208,6 +217,30 @@ public FlinkInputFormat buildFormat() {
.transform(readerOperatorName, typeInfo, StreamingReaderOperator.factory(format));
}
}
+
+ private DataStream<RowData> createInputDataStream(FlinkInputFormat format, ScanContext context,
+ TypeInformation<RowData> typeInfo) {
+ int parallelism = flinkConf.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
+ if (flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
+ int max = flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
Review comment:
Nit: use maxInterParallelism pls.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -208,6 +217,30 @@ public FlinkInputFormat buildFormat() {
.transform(readerOperatorName, typeInfo, StreamingReaderOperator.factory(format));
}
}
+
+ private DataStream<RowData> createInputDataStream(FlinkInputFormat format, ScanContext context,
+ TypeInformation<RowData> typeInfo) {
+ int parallelism = flinkConf.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
+ if (flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
+ int max = flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
+ Preconditions.checkState(max >= 1,
+ FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX.key() + " cannot be less than 1");
+ int splitNum = 0;
Review comment:
Nit: this assignment is redundant ( from intellij).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#issuecomment-761290558
@openinx, is this something we should try to get into the 0.11.0 release?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r562446908
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -106,6 +109,51 @@ public void testResiduals() throws Exception {
assertRecords(runWithFilter(filter, "where dt='2020-03-20' and id=123"), expectedRecords, SCHEMA);
}
+ @Test
+ public void testInferedParallelism() throws IOException {
+ Table table = catalog.createTable(TableIdentifier.of("default", "t"), SCHEMA, SPEC);
+
+ TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
+ FlinkInputFormat flinkInputFormat = FlinkSource.forRowData().tableLoader(tableLoader).table(table).buildFormat();
+ ScanContext scanContext = ScanContext.builder().build();
+
+ // Empty table ,parallelism at least 1
+ int parallelism = FlinkSource.forRowData()
+ .flinkConf(new Configuration())
+ .inferParallelism(flinkInputFormat, scanContext);
+ Assert.assertEquals("Should produce the expected parallelism.", 1, parallelism);
+
+ List<Record> writeRecords = RandomGenericData.generate(SCHEMA, 2, 0L);
+ writeRecords.get(0).set(1, 123L);
+ writeRecords.get(0).set(2, "2020-03-20");
+ writeRecords.get(1).set(1, 456L);
+ writeRecords.get(1).set(2, "2020-03-20");
+
+ GenericAppenderHelper helper = new GenericAppenderHelper(table, fileFormat, TEMPORARY_FOLDER);
+
+ DataFile dataFile1 = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0), writeRecords);
+ DataFile dataFile2 = helper.writeFile(TestHelpers.Row.of("2020-03-21", 0),
+ RandomGenericData.generate(SCHEMA, 2, 0L));
+ helper.appendToTable(dataFile1, dataFile2);
+
+ // Make sure to generate 2 CombinedScanTasks
+ long maxFileLen = Math.max(dataFile1.fileSizeInBytes(), dataFile2.fileSizeInBytes());
+ executeSQL(String
+ .format("ALTER TABLE t SET ('read.split.open-file-cost'='1', 'read.split.target-size'='%s')", maxFileLen));
+
+ // 2 splits ,the parallelism is 2
+ parallelism = FlinkSource.forRowData()
Review comment:
I add the test case , but I did not split these test cases into different methods because they share a lot of code. If they are split, there may be a lot of duplicate code.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on pull request #1936: Flink : add parallelism optimize for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#issuecomment-745238862
@openinx could you help review this pr when you have time , thanks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r561502241
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -70,6 +73,7 @@ public static Builder forRowData() {
private Table table;
private TableLoader tableLoader;
private TableSchema projectedSchema;
+ private ReadableConfig flinkConf;
Review comment:
When we construct the `IcebergTableSource`, we use the `TableSourceFactory.Context#getConfiguration` method to get the configuration. This method returns a ReadableConfig, so we use ReadableConfig instead of Configuration. In addition, Configuration is the implementation class of the ReadableConfig interface, so I think ReadableConfig should not has a problem
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r561515156
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
> Just as you commented that it is best not to use the internal code of flink, I think we should try to use public APIs to get information.
The comment that saying we'd better not use flink's Internal API because that would introduce extra upgrade complexity (new flink version may breaks those internal API so we iceberg have to adjust the codes, finally maintaining different versions of flink will bring us a lot of burden).
Writing iceberg unit tests based on our iceberg's non-public ( we usually use package-access ) method is OK because there's no extra burden .
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r561531089
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -70,6 +73,7 @@ public static Builder forRowData() {
private Table table;
private TableLoader tableLoader;
private TableSchema projectedSchema;
+ private ReadableConfig flinkConf;
Review comment:
Got your point, I'd prefer to use flink's `Configuration` because it will be exposed to flink developers as an API in `FlinkSource`, using the unified `Configuration` will be more straightforward for them. But as you said, the TableSourceFactory#Context is exposing the `ReadableConfig`, I also did not find a correct way to convert `ReadableConfig` to `Configuration`. OK, I think we could use `ReadableConfig` here, thanks.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r560864790
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
Although we use streaming mode, we read a batch of data, so flink will treat this batch job as a bounded stream. I think whether we should use `StreamExecutionEnvironment` to process batch jobs and streaming jobs at the same time?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #1936: Flink: Support inferring parallelism for batch read.
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#issuecomment-765301876
Got this merged, thanks @zhangjun0x01 for contributing !
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r561636616
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
I update the pr,move the `testInferedParallelism` method to TestFlinkScanSql,use `FlinkSource.Builder#inferParallelism` method to do the assertion
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r555468396
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -103,4 +124,45 @@ public void testLimitPushDown() {
Assert.assertEquals("should have 1 record", 1, mixedResult.size());
Assert.assertArrayEquals("Should produce the expected records", mixedResult.get(0), new Object[] {1, "a"});
}
+
+ @Test
+ public void testParallelismOptimize() {
+ sql("INSERT INTO %s VALUES (1,'hello')", TABLE_NAME);
+ sql("INSERT INTO %s VALUES (2,'iceberg')", TABLE_NAME);
+
+ TableEnvironment tenv = getTableEnv();
+
+ // empty table ,parallelism at least 1
+ Table tableEmpty = tenv.sqlQuery(String.format("SELECT * FROM %s", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(1, tableEmpty, tenv);
+
+ // make sure to generate 2 CombinedScanTasks
+ org.apache.iceberg.Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
+ Stream<FileScanTask> stream = StreamSupport.stream(table.newScan().planFiles().spliterator(), false);
+ Optional<FileScanTask> fileScanTaskOptional = stream.max(Comparator.comparing(FileScanTask::length));
+ Assert.assertTrue(fileScanTaskOptional.isPresent());
+ long maxFileLen = fileScanTaskOptional.get().length();
+ sql("ALTER TABLE %s SET ('read.split.open-file-cost'='1', 'read.split.target-size'='%s')", TABLE_NAME, maxFileLen);
+
+ // 2 splits ,the parallelism is 2
+ Table tableSelect = tenv.sqlQuery(String.format("SELECT * FROM %s", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(2, tableSelect, tenv);
+
+ // 2 splits and limit is 1 ,the parallelism is 1
+ Table tableLimit = tenv.sqlQuery(String.format("SELECT * FROM %s LIMIT 1", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(1, tableLimit, tenv);
+ }
+
+ private void testParallelismSettingTranslateAndAssert(int expected, Table table, TableEnvironment tEnv) {
Review comment:
Here we need to get the parallelism of an Operator. It is indeed get by using the flink internal code, because I have not found other ways to obtain the parallelism.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r561461670
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
I think these are two different concepts.
In flink, whether using batch mode or streaming mode, we can read batch data. Flink treats batch jobs as bounded streaming jobs, so there should be no problem whether it is using batch mode or streaming mode to read batch data. In addition, flink will use `StreamExecutionEnvironment (DataStream)` to do batch tasks and stream tasks uniformly ([the doc link](https://flink.apache.org/news/2020/12/10/release-1.12.0.html#batch-execution-mode-in-the-datastream-api)) . The batch mode may expire, so I think we should also use `StreamExecutionEnvironment (DataStream)` for batch tasks as much as possible.
When we use `StreamExecutionEnvironment` , in `FlinkSource.Builder#build` method,
the `if` and `else` block in this method are both streaming jobs, `if` block is a bounded streaming jobs, maybe we can rename `ScanContext#isStreaming` field to `ScanContext#isStreamingRead`, which will be easier to understand. `If` code block is bounded stream job (batch), `else` code block to do long-running stream job.
```
if (!context.isStreaming()) {
int parallelism = inferParallelism(format, context);
return env.createInput(format, typeInfo).setParallelism(parallelism);
} else {
StreamingMonitorFunction function = new StreamingMonitorFunction(tableLoader, context);
String monitorFunctionName = String.format("Iceberg table (%s) monitor", table);
String readerOperatorName = String.format("Iceberg table (%s) reader", table);
return env.addSource(function, monitorFunctionName)
.transform(readerOperatorName, typeInfo, StreamingReaderOperator.factory(format));
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#issuecomment-761999393
According to @rdblue [suggestion](https://github.com/apache/iceberg/pull/1893#issuecomment-761613165), I refactored this class, using only `HadoopCatalog` and Avro format
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r562373214
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -106,6 +109,51 @@ public void testResiduals() throws Exception {
assertRecords(runWithFilter(filter, "where dt='2020-03-20' and id=123"), expectedRecords, SCHEMA);
}
+ @Test
+ public void testInferedParallelism() throws IOException {
+ Table table = catalog.createTable(TableIdentifier.of("default", "t"), SCHEMA, SPEC);
+
+ TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
+ FlinkInputFormat flinkInputFormat = FlinkSource.forRowData().tableLoader(tableLoader).table(table).buildFormat();
+ ScanContext scanContext = ScanContext.builder().build();
+
+ // Empty table ,parallelism at least 1
Review comment:
Nit: `inter parallelism should be at least 1.`
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -208,6 +218,33 @@ public FlinkInputFormat buildFormat() {
.transform(readerOperatorName, typeInfo, StreamingReaderOperator.factory(format));
}
}
+
+ int inferParallelism(FlinkInputFormat format, ScanContext context) {
+ int parallelism = readableConfig.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
+ if (readableConfig.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
+ int maxInterParallelism = readableConfig.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
Review comment:
Nit: `maxInterParallelism` -> `maxInferParallelism`, seems like it's a typo ?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -79,13 +84,20 @@ public boolean isBounded() {
@Override
public TableSource<RowData> projectFields(int[] fields) {
- return new IcebergTableSource(loader, schema, properties, fields, isLimitPushDown, limit, filters);
+ return new IcebergTableSource(loader, schema, properties, fields, isLimitPushDown, limit, filters, readableConfig);
}
@Override
public DataStream<RowData> getDataStream(StreamExecutionEnvironment execEnv) {
- return FlinkSource.forRowData().env(execEnv).tableLoader(loader).project(getProjectedSchema()).limit(limit)
- .filters(filters).properties(properties).build();
+ return FlinkSource.forRowData()
Review comment:
Nit: maybe we'd better also align the orders as above commented.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -70,6 +73,7 @@ public static Builder forRowData() {
private Table table;
private TableLoader tableLoader;
private TableSchema projectedSchema;
+ private ReadableConfig readableConfig;
Review comment:
Should we provide a `new Configuration()` for this variable ? Otherwise, it will just throw NPE if people forget to provide a `flinkConf` in FlinkSource#Builder because we don't check the nullable in `interParallelism`.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -106,6 +109,51 @@ public void testResiduals() throws Exception {
assertRecords(runWithFilter(filter, "where dt='2020-03-20' and id=123"), expectedRecords, SCHEMA);
}
+ @Test
+ public void testInferedParallelism() throws IOException {
+ Table table = catalog.createTable(TableIdentifier.of("default", "t"), SCHEMA, SPEC);
+
+ TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
+ FlinkInputFormat flinkInputFormat = FlinkSource.forRowData().tableLoader(tableLoader).table(table).buildFormat();
+ ScanContext scanContext = ScanContext.builder().build();
+
+ // Empty table ,parallelism at least 1
+ int parallelism = FlinkSource.forRowData()
+ .flinkConf(new Configuration())
+ .inferParallelism(flinkInputFormat, scanContext);
+ Assert.assertEquals("Should produce the expected parallelism.", 1, parallelism);
+
+ List<Record> writeRecords = RandomGenericData.generate(SCHEMA, 2, 0L);
+ writeRecords.get(0).set(1, 123L);
+ writeRecords.get(0).set(2, "2020-03-20");
+ writeRecords.get(1).set(1, 456L);
+ writeRecords.get(1).set(2, "2020-03-20");
+
+ GenericAppenderHelper helper = new GenericAppenderHelper(table, fileFormat, TEMPORARY_FOLDER);
+
+ DataFile dataFile1 = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0), writeRecords);
+ DataFile dataFile2 = helper.writeFile(TestHelpers.Row.of("2020-03-21", 0),
+ RandomGenericData.generate(SCHEMA, 2, 0L));
+ helper.appendToTable(dataFile1, dataFile2);
+
+ // Make sure to generate 2 CombinedScanTasks
+ long maxFileLen = Math.max(dataFile1.fileSizeInBytes(), dataFile2.fileSizeInBytes());
+ executeSQL(String
+ .format("ALTER TABLE t SET ('read.split.open-file-cost'='1', 'read.split.target-size'='%s')", maxFileLen));
+
+ // 2 splits ,the parallelism is 2
+ parallelism = FlinkSource.forRowData()
Review comment:
I think there're other test cases that we don't cover, it's good to cover those tests.
1. table.exec.iceberg.infer-source-parallelism=false;
2. table.exec.iceberg.infer-source-parallelism.max <= numberOfSplits;
3. table.exec.iceberg.infer-source-parallelism.max > numberOfSplits;
4. table.exec.iceberg.infer-source-parallelism.max > limit;
5. table.exec.iceberg.infer-source-parallelism.max <= limit;
Divide those cases into small method if necessary.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -51,25 +52,29 @@
private final TableLoader loader;
private final TableSchema schema;
private final Map<String, String> properties;
+ private final ReadableConfig readableConfig;
Review comment:
Nit: Let's move this line to line60, so that the assignment order of `IcebergTableSource` constructor could align with these definitions.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
##########
@@ -106,6 +109,51 @@ public void testResiduals() throws Exception {
assertRecords(runWithFilter(filter, "where dt='2020-03-20' and id=123"), expectedRecords, SCHEMA);
}
+ @Test
+ public void testInferedParallelism() throws IOException {
+ Table table = catalog.createTable(TableIdentifier.of("default", "t"), SCHEMA, SPEC);
+
+ TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
+ FlinkInputFormat flinkInputFormat = FlinkSource.forRowData().tableLoader(tableLoader).table(table).buildFormat();
+ ScanContext scanContext = ScanContext.builder().build();
+
+ // Empty table ,parallelism at least 1
+ int parallelism = FlinkSource.forRowData()
+ .flinkConf(new Configuration())
+ .inferParallelism(flinkInputFormat, scanContext);
+ Assert.assertEquals("Should produce the expected parallelism.", 1, parallelism);
+
+ List<Record> writeRecords = RandomGenericData.generate(SCHEMA, 2, 0L);
+ writeRecords.get(0).set(1, 123L);
+ writeRecords.get(0).set(2, "2020-03-20");
+ writeRecords.get(1).set(1, 456L);
+ writeRecords.get(1).set(2, "2020-03-20");
+
+ GenericAppenderHelper helper = new GenericAppenderHelper(table, fileFormat, TEMPORARY_FOLDER);
+
+ DataFile dataFile1 = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0), writeRecords);
+ DataFile dataFile2 = helper.writeFile(TestHelpers.Row.of("2020-03-21", 0),
+ RandomGenericData.generate(SCHEMA, 2, 0L));
Review comment:
Those random generated records will be located in partition `2020-03-21` ? I guess it's not.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -137,7 +136,10 @@ public void testFilterPushDownEqual() {
Assert.assertEquals("Should have 1 record", 1, result.size());
Assert.assertArrayEquals("Should produce the expected record", expectRecord, result.get(0));
- Assert.assertEquals("Should create only one scan", 1, scanEventCount);
+ // Because we add infer parallelism, all data files will be scanned first.
+ // Flink will call FlinkInputFormat#createInputSplits method to scan the data files,
+ // plus the operation to get the execution plan, so there are three scan event.
+ Assert.assertEquals("Should create 3 scans", 3, scanEventCount);
Review comment:
We can disable the `table.exec.iceberg.infer-source-parallelism` for all the batch tests by default, then we don't have to change all cases from this file. Actually, we have wrote many unit tests which depends on the `parallelism`, for example this PR https://github.com/apache/iceberg/pull/2064. Using the inter-parallelism for batch unit tests will introduce extra complexity and instability, so I recommend to disable the infer parallelism in our batch unit tests by default:
```diff
diff --git a/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java b/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
index 5b8e58cf..ab3d56ea 100644
--- a/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
+++ b/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
@@ -62,10 +62,17 @@ public abstract class FlinkTestBase extends AbstractTestBase {
if (tEnv == null) {
synchronized (this) {
if (tEnv == null) {
- this.tEnv = TableEnvironment.create(EnvironmentSettings
+ EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.useBlinkPlanner()
- .inBatchMode().build());
+ .inBatchMode()
+ .build();
+
+ TableEnvironment env = TableEnvironment.create(settings);
+ env.getConfig().getConfiguration()
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+
+ tEnv = env;
}
}
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r559301071
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -103,4 +124,45 @@ public void testLimitPushDown() {
Assert.assertEquals("should have 1 record", 1, mixedResult.size());
Assert.assertArrayEquals("Should produce the expected records", mixedResult.get(0), new Object[] {1, "a"});
}
+
+ @Test
+ public void testParallelismOptimize() {
+ sql("INSERT INTO %s VALUES (1,'hello')", TABLE_NAME);
+ sql("INSERT INTO %s VALUES (2,'iceberg')", TABLE_NAME);
+
+ TableEnvironment tenv = getTableEnv();
+
+ // empty table ,parallelism at least 1
+ Table tableEmpty = tenv.sqlQuery(String.format("SELECT * FROM %s", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(1, tableEmpty, tenv);
+
+ // make sure to generate 2 CombinedScanTasks
+ org.apache.iceberg.Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
+ Stream<FileScanTask> stream = StreamSupport.stream(table.newScan().planFiles().spliterator(), false);
+ Optional<FileScanTask> fileScanTaskOptional = stream.max(Comparator.comparing(FileScanTask::length));
+ Assert.assertTrue(fileScanTaskOptional.isPresent());
+ long maxFileLen = fileScanTaskOptional.get().length();
+ sql("ALTER TABLE %s SET ('read.split.open-file-cost'='1', 'read.split.target-size'='%s')", TABLE_NAME, maxFileLen);
+
+ // 2 splits ,the parallelism is 2
+ Table tableSelect = tenv.sqlQuery(String.format("SELECT * FROM %s", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(2, tableSelect, tenv);
+
+ // 2 splits and limit is 1 ,the parallelism is 1
+ Table tableLimit = tenv.sqlQuery(String.format("SELECT * FROM %s LIMIT 1", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(1, tableLimit, tenv);
+ }
+
+ private void testParallelismSettingTranslateAndAssert(int expected, Table table, TableEnvironment tEnv) {
Review comment:
Flink 1.12 does refactor `ExecNode`, I found an easier way to assert parallelism, I will update it later
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -103,4 +124,45 @@ public void testLimitPushDown() {
Assert.assertEquals("should have 1 record", 1, mixedResult.size());
Assert.assertArrayEquals("Should produce the expected records", mixedResult.get(0), new Object[] {1, "a"});
}
+
+ @Test
+ public void testParallelismOptimize() {
+ sql("INSERT INTO %s VALUES (1,'hello')", TABLE_NAME);
+ sql("INSERT INTO %s VALUES (2,'iceberg')", TABLE_NAME);
+
+ TableEnvironment tenv = getTableEnv();
+
+ // empty table ,parallelism at least 1
+ Table tableEmpty = tenv.sqlQuery(String.format("SELECT * FROM %s", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(1, tableEmpty, tenv);
+
+ // make sure to generate 2 CombinedScanTasks
+ org.apache.iceberg.Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
+ Stream<FileScanTask> stream = StreamSupport.stream(table.newScan().planFiles().spliterator(), false);
+ Optional<FileScanTask> fileScanTaskOptional = stream.max(Comparator.comparing(FileScanTask::length));
+ Assert.assertTrue(fileScanTaskOptional.isPresent());
+ long maxFileLen = fileScanTaskOptional.get().length();
+ sql("ALTER TABLE %s SET ('read.split.open-file-cost'='1', 'read.split.target-size'='%s')", TABLE_NAME, maxFileLen);
+
+ // 2 splits ,the parallelism is 2
+ Table tableSelect = tenv.sqlQuery(String.format("SELECT * FROM %s", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(2, tableSelect, tenv);
+
+ // 2 splits and limit is 1 ,the parallelism is 1
+ Table tableLimit = tenv.sqlQuery(String.format("SELECT * FROM %s LIMIT 1", TABLE_NAME));
+ testParallelismSettingTranslateAndAssert(1, tableLimit, tenv);
+ }
+
+ private void testParallelismSettingTranslateAndAssert(int expected, Table table, TableEnvironment tEnv) {
Review comment:
Here we need to get the parallelism of an Operator. It is indeed get by using the flink internal code, because I have not found other ways to obtain the parallelism.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r560992378
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
If someone provides both flink `Configuration` and iceberg's `properties`, then we should use the flink's `Configuration` values overwrite the iceberg's properties because `properties` is a table-level settings while the flink's `Configuration` is a job-level settings. It is reasonable for fine-grained configuration to ovewrite coarse-grained configuration.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r562450473
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -137,7 +136,10 @@ public void testFilterPushDownEqual() {
Assert.assertEquals("Should have 1 record", 1, result.size());
Assert.assertArrayEquals("Should produce the expected record", expectRecord, result.get(0));
- Assert.assertEquals("Should create only one scan", 1, scanEventCount);
+ // Because we add infer parallelism, all data files will be scanned first.
+ // Flink will call FlinkInputFormat#createInputSplits method to scan the data files,
+ // plus the operation to get the execution plan, so there are three scan event.
+ Assert.assertEquals("Should create 3 scans", 3, scanEventCount);
Review comment:
yes,I update 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r561506300
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
The `isStreaming` indicate whether the flink source is a streaming source (In our mind) , not say it's a streaming job or batch job. The hive table source also has the similar configure key :
```java
public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
key("streaming-source.enable")
.booleanType()
.defaultValue(false)
.withDescription(
"Enable streaming source or not.\n"
+ " NOTES: Please make sure that each partition/file should be written"
+ " atomically, otherwise the reader may get incomplete data.");
```
If we think this iceberg configure key is not very clear, I think we could propose another separate PR to align with hive configure key. Let's focus on this parallelism issue here, what do you think ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r555467464
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
##########
@@ -195,7 +205,29 @@ public FlinkInputFormat buildFormat() {
Preconditions.checkNotNull(env, "StreamExecutionEnvironment should not be null");
FlinkInputFormat format = buildFormat();
if (isBounded(context)) {
- return env.createInput(format, rowTypeInfo);
+ int parallelism = flinkConf.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
+ if (flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
+ int max = flinkConf.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
+ if (max < 1) {
+ throw new IllegalConfigurationException(
+ FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX.key() + " cannot be less than 1");
+ }
+
+ int splitNum = 0;
+ try {
+ FlinkInputSplit[] splits = format.createInputSplits(0);
+ splitNum = splits.length;
+ } catch (IOException e) {
+ throw new RuntimeException("get input split error.", e);
+ }
+
+ parallelism = Math.min(splitNum, max);
+ }
+
+ parallelism = limit > 0 ? Math.min(parallelism, (int) limit) : parallelism;
Review comment:
the `parallelism` is int type and the `limit` is long type, `Math.min(parallelism, limit)` will throws an exception,I add a judgment to prevent overflow.
```
int limitInt = limit > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) limit;
parallelism = limitInt > 0 ? Math.min(parallelism, limitInt) : parallelism;
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1936: Flink: Default read parallelism to number of splits
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1936:
URL: https://github.com/apache/iceberg/pull/1936#discussion_r560864790
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -685,4 +782,60 @@ public void testSqlParseError() {
AssertHelpers.assertThrows("The NaN is not supported by flink now. ",
NumberFormatException.class, () -> sql(sqlParseErrorLTE));
}
+
+ /**
+ * The sql can be executed in both streaming and batch mode, in order to get the parallelism, we convert the flink
+ * Table to flink DataStream, so we only use streaming mode here.
+ *
+ * @throws TableNotExistException table not exist exception
+ */
+ @Test
+ public void testInferedParallelism() throws TableNotExistException {
+ Assume.assumeTrue("The execute mode should be streaming mode", isStreamingJob);
Review comment:
Although we use streaming mode, we read a batch of data, so flink will treat this batch job as a bounded stream. I think whether we should use `StreamExecutionEnvironment` to process batch jobs and streaming jobs at the same time?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org