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:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
