You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2021/07/15 11:41:06 UTC
[flink] branch release-1.13 updated: [FLINK-22898][Connectors/Hive]
return default-parallelism in HiveParallelismInference when there is no
source infer
This is an automated email from the ASF dual-hosted git repository.
lirui pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.13 by this push:
new 004d3b5 [FLINK-22898][Connectors/Hive] return default-parallelism in HiveParallelismInference when there is no source infer
004d3b5 is described below
commit 004d3b58e4f4637efb09e3c7c84f01c3bd64ad9f
Author: yuxia Luo <lu...@alibaba-inc.com>
AuthorDate: Thu Jul 8 15:54:24 2021 +0800
[FLINK-22898][Connectors/Hive] return default-parallelism in HiveParallelismInference when there is no source infer
This closes #16421
---
.../connectors/hive/HiveParallelismInference.java | 4 ++
.../connectors/hive/HiveTableSourceITCase.java | 46 +++++++++++++++++++++-
2 files changed, 49 insertions(+), 1 deletion(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveParallelismInference.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveParallelismInference.java
index 880480c..850008f 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveParallelismInference.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveParallelismInference.java
@@ -59,6 +59,10 @@ class HiveParallelismInference {
* SELECT * FROM xxx LIMIT [limit]</code>.
*/
int limit(Long limit) {
+ if (!infer) {
+ return parallelism;
+ }
+
if (limit != null) {
parallelism = Math.min(parallelism, (int) (limit / 1000));
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
index 2161f5f..138b9fd 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
@@ -24,6 +24,7 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.HiveVersionTestUtil;
+import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
@@ -524,7 +525,50 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
.get(0);
Transformation<?> transformation =
(execNode.translateToPlan(planner).getInputs().get(0)).getInputs().get(0);
- Assert.assertEquals(1, transformation.getParallelism());
+ // when there's no infer, should use the default parallelism configured
+ Assert.assertEquals(2, transformation.getParallelism());
+ }
+
+ @Test
+ public void testParallelismWithoutParallelismInfer() throws Exception {
+ final String dbName = "source_db";
+ final String tblName = "test_parallelism_no_infer";
+ EnvironmentSettings settings =
+ EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
+ TableEnvironment tEnv = TableEnvironment.create(settings);
+ tEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+ tEnv.registerCatalog("hive", hiveCatalog);
+ tEnv.useCatalog("hive");
+ tEnv.getConfig()
+ .getConfiguration()
+ .setBoolean(HiveOptions.TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM, false);
+ tEnv.executeSql(
+ "CREATE TABLE source_db.test_parallelism_no_infer "
+ + "(`year` STRING, `value` INT) partitioned by (pt int)");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
+ .addRow(new Object[] {"2014", 3})
+ .addRow(new Object[] {"2014", 4})
+ .commit("pt=0");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
+ .addRow(new Object[] {"2015", 2})
+ .addRow(new Object[] {"2015", 5})
+ .commit("pt=1");
+ Table table =
+ tEnv.sqlQuery("select * from hive.source_db.test_parallelism_no_infer limit 1");
+ PlannerBase planner = (PlannerBase) ((TableEnvironmentImpl) tEnv).getPlanner();
+ RelNode relNode = planner.optimize(TableTestUtil.toRelNode(table));
+ ExecNode<?> execNode =
+ planner.translateToExecNodeGraph(toScala(Collections.singletonList(relNode)))
+ .getRootNodes()
+ .get(0);
+ Transformation<?> transformation =
+ (execNode.translateToPlan(planner).getInputs().get(0)).getInputs().get(0);
+ // when there's no infer, should use the default parallelism
+ Assert.assertEquals(
+ ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM
+ .defaultValue()
+ .intValue(),
+ transformation.getParallelism());
}
@Test