luoyuxia commented on code in PR #21703:
URL: https://github.com/apache/flink/pull/21703#discussion_r1080818665
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java:
##########
@@ -351,15 +365,211 @@ private DataStreamSink<?> consume(
}
}
- private DataStreamSink<Row> createBatchSink(
+ private DataStreamSink<?> createBatchSink(
+ DataStream<RowData> dataStream,
+ DataStructureConverter converter,
+ HiveWriterFactory recordWriterFactory,
+ TableMetaStoreFactory metaStoreFactory,
+ OutputFileConfig.OutputFileConfigBuilder fileConfigBuilder,
+ String stagingParentDir,
+ StorageDescriptor sd,
+ Properties tableProps,
+ boolean isToLocal,
+ boolean overwrite,
+ int sinkParallelism)
+ throws IOException {
+ org.apache.flink.configuration.Configuration conf =
+ new org.apache.flink.configuration.Configuration();
+ catalogTable.getOptions().forEach(conf::setString);
+ boolean autoCompaction =
conf.getBoolean(FileSystemConnectorOptions.AUTO_COMPACTION);
+ if (autoCompaction) {
+ if (batchShuffleMode != BatchShuffleMode.ALL_EXCHANGES_BLOCKING) {
Review Comment:
Thanks for configuring it out. After think it over, yes, it should still
work in pipeline shufle mode.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
