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/08 03:17:08 UTC
[flink] branch master updated: [FLINK-23178][hive] Raise an error
for writing stream data into partitioned hive tables without a partition
committer
This is an automated email from the ASF dual-hosted git repository.
lirui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new a94745e [FLINK-23178][hive] Raise an error for writing stream data into partitioned hive tables without a partition committer
a94745e is described below
commit a94745ec85bf9e8ca3bc2fced5c1a466b836e0be
Author: Rui Li <li...@apache.org>
AuthorDate: Mon Jul 5 15:31:30 2021 +0800
[FLINK-23178][hive] Raise an error for writing stream data into partitioned hive tables without a partition committer
This closes #16370
---
.../flink/connectors/hive/HiveTableSink.java | 21 ++++++++-----
.../flink/connectors/hive/HiveTableSinkITCase.java | 35 ++++++++++++++++++++++
2 files changed, 48 insertions(+), 8 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java
index 138befc..b31da9b 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.common.serialization.BulkWriter;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connectors.hive.read.HiveCompactReaderFactory;
import org.apache.flink.connectors.hive.util.HiveConfUtils;
+import org.apache.flink.connectors.hive.util.JobConfUtils;
import org.apache.flink.connectors.hive.write.HiveBulkWriterFactory;
import org.apache.flink.connectors.hive.write.HiveOutputFormatFactory;
import org.apache.flink.connectors.hive.write.HiveWriterFactory;
@@ -67,6 +68,7 @@ import org.apache.flink.table.utils.TableSchemaUtils;
import org.apache.flink.types.Row;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -217,7 +219,7 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
builder.setPartitionComputer(
new HiveRowPartitionComputer(
hiveShim,
- defaultPartName(),
+ JobConfUtils.getDefaultPartitionName(jobConf),
tableSchema.getFieldNames(),
tableSchema.getFieldDataTypes(),
getPartitionKeyArray()));
@@ -248,10 +250,19 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
new org.apache.flink.configuration.Configuration();
catalogTable.getOptions().forEach(conf::setString);
+ String commitPolicies = conf.getString(FileSystemOptions.SINK_PARTITION_COMMIT_POLICY_KIND);
+ if (!getPartitionKeys().isEmpty() && StringUtils.isNullOrWhitespaceOnly(commitPolicies)) {
+ throw new FlinkHiveException(
+ String.format(
+ "Streaming write to partitioned hive table %s without providing a commit policy. "
+ + "Make sure to set a proper value for %s",
+ identifier, FileSystemOptions.SINK_PARTITION_COMMIT_POLICY_KIND.key()));
+ }
+
HiveRowDataPartitionComputer partComputer =
new HiveRowDataPartitionComputer(
hiveShim,
- defaultPartName(),
+ JobConfUtils.getDefaultPartitionName(jobConf),
tableSchema.getFieldNames(),
tableSchema.getFieldDataTypes(),
getPartitionKeyArray());
@@ -332,12 +343,6 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
writerStream, path, identifier, getPartitionKeys(), msFactory(), fsFactory(), conf);
}
- private String defaultPartName() {
- return jobConf.get(
- HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
- HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
- }
-
private CompactReader.Factory<RowData> createCompactReaderFactory(
StorageDescriptor sd, Properties properties) {
return new HiveCompactReaderFactory(
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
index 4ae3a07..5f2e0e4 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
@@ -65,6 +65,8 @@ import static org.apache.flink.table.planner.utils.TableTestUtil.readFromResourc
import static org.apache.flink.table.planner.utils.TableTestUtil.replaceStageId;
import static org.apache.flink.table.planner.utils.TableTestUtil.replaceStreamNodeId;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
/** Tests {@link HiveTableSink}. */
public class HiveTableSinkITCase {
@@ -375,6 +377,39 @@ public class HiveTableSinkITCase {
}
}
+ @Test
+ public void testStreamingSinkWithoutCommitPolicy() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tableEnv = HiveTestUtils.createTableEnvInStreamingMode(env);
+ tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
+ tableEnv.useCatalog(hiveCatalog.getName());
+
+ tableEnv.executeSql("create database db1");
+ try {
+ tableEnv.useDatabase("db1");
+ tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+ tableEnv.executeSql("create table dest(x int) partitioned by (p string)");
+
+ tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
+ tableEnv.executeSql(
+ "create table src (i int, p string) with ("
+ + "'connector'='datagen',"
+ + "'number-of-rows'='5')");
+ tableEnv.executeSql("insert into dest select * from src").await();
+ fail("Streaming write partitioned table without commit policy should fail");
+ } catch (FlinkHiveException e) {
+ // expected
+ assertTrue(
+ e.getMessage()
+ .contains(
+ String.format(
+ "Streaming write to partitioned hive table `%s`.`%s`.`%s` without providing a commit policy",
+ hiveCatalog.getName(), "db1", "dest")));
+ } finally {
+ tableEnv.executeSql("drop database db1 cascade");
+ }
+ }
+
private static List<String> fetchRows(Iterator<Row> iter, int size) {
List<String> strings = new ArrayList<>(size);
for (int i = 0; i < size; i++) {