You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2022/03/22 19:13:49 UTC
[flink] 01/08: [hotfix][connector-hive] Avoid serializing TableConfig
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0ba337cd9f1f6f374aba662769c0ff2b2013e49e
Author: Marios Trivyzas <ma...@gmail.com>
AuthorDate: Mon Mar 21 13:49:49 2022 +0200
[hotfix][connector-hive] Avoid serializing TableConfig
Use and pass around only `threadNum` which is the only option read,
instead of the whole `TableConfig`, to prevent the relevant classes
that are serialized from trying to serialize also the `TableConfig`.
(cherry picked from commit 7fcd6935e065bb0b8dd0b7992ffb2887dab5a5f8)
---
.../hive/ContinuousHiveSplitEnumerator.java | 13 ++++-----
.../connectors/hive/HiveLookupTableSource.java | 2 +-
.../apache/flink/connectors/hive/HiveSource.java | 11 ++++----
.../flink/connectors/hive/HiveSourceBuilder.java | 18 ++++++++----
.../connectors/hive/HiveSourceFileEnumerator.java | 29 +++++++------------
.../flink/connectors/hive/HiveTableSink.java | 33 ++++++++++++++++++----
.../flink/connectors/hive/HiveTableSource.java | 4 ++-
.../hive/read/HiveInputFormatPartitionReader.java | 9 +++---
.../connectors/hive/read/HiveTableInputFormat.java | 20 ++++---------
.../connectors/hive/PartitionMonitorTest.java | 3 +-
.../read/HiveInputFormatPartitionReaderITCase.java | 4 +--
11 files changed, 79 insertions(+), 67 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/ContinuousHiveSplitEnumerator.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/ContinuousHiveSplitEnumerator.java
index 983faac..a061b27 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/ContinuousHiveSplitEnumerator.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/ContinuousHiveSplitEnumerator.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.PendingSplitsCheckpoint;
import org.apache.flink.connector.file.src.assigners.FileSplitAssigner;
@@ -78,7 +77,7 @@ public class ContinuousHiveSplitEnumerator<T extends Comparable<T>>
Collection<List<String>> seenPartitionsSinceOffset,
FileSplitAssigner splitAssigner,
long discoveryInterval,
- ReadableConfig flinkConf,
+ int threadNum,
JobConf jobConf,
ObjectPath tablePath,
ContinuousPartitionFetcher<Partition, T> fetcher,
@@ -95,7 +94,7 @@ public class ContinuousHiveSplitEnumerator<T extends Comparable<T>>
currentReadOffset,
seenPartitionsSinceOffset,
tablePath,
- flinkConf,
+ threadNum,
jobConf,
fetcher,
fetcherContext);
@@ -188,7 +187,7 @@ public class ContinuousHiveSplitEnumerator<T extends Comparable<T>>
private final Set<List<String>> seenPartitionsSinceOffset;
private final ObjectPath tablePath;
- private final ReadableConfig flinkConf;
+ private final int threadNum;
private final JobConf jobConf;
private final ContinuousPartitionFetcher<Partition, T> fetcher;
private final HiveContinuousPartitionContext<Partition, T> fetcherContext;
@@ -197,14 +196,14 @@ public class ContinuousHiveSplitEnumerator<T extends Comparable<T>>
T currentReadOffset,
Collection<List<String>> seenPartitionsSinceOffset,
ObjectPath tablePath,
- ReadableConfig flinkConf,
+ int threadNum,
JobConf jobConf,
ContinuousPartitionFetcher<Partition, T> fetcher,
HiveContinuousPartitionContext<Partition, T> fetcherContext) {
this.currentReadOffset = currentReadOffset;
this.seenPartitionsSinceOffset = new HashSet<>(seenPartitionsSinceOffset);
this.tablePath = tablePath;
- this.flinkConf = flinkConf;
+ this.threadNum = threadNum;
this.jobConf = jobConf;
this.fetcher = fetcher;
this.fetcherContext = fetcherContext;
@@ -244,7 +243,7 @@ public class ContinuousHiveSplitEnumerator<T extends Comparable<T>>
0,
Collections.singletonList(
fetcherContext.toHiveTablePartition(partition)),
- flinkConf,
+ threadNum,
jobConf));
}
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveLookupTableSource.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveLookupTableSource.java
index 488ea1b..e68072a 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveLookupTableSource.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveLookupTableSource.java
@@ -242,7 +242,7 @@ public class HiveLookupTableSource extends HiveTableSource implements LookupTabl
PartitionReader<HiveTablePartition, RowData> partitionReader =
new HiveInputFormatPartitionReader(
- flinkConf,
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM),
jobConf,
hiveVersion,
tablePath,
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSource.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSource.java
index 26c4ac5..d7f5209 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSource.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSource.java
@@ -22,7 +22,6 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
-import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.file.src.AbstractFileSource;
import org.apache.flink.connector.file.src.ContinuousEnumerationSettings;
import org.apache.flink.connector.file.src.PendingSplitsCheckpoint;
@@ -60,7 +59,7 @@ public class HiveSource<T> extends AbstractFileSource<T, HiveSourceSplit> {
private static final long serialVersionUID = 1L;
- private final ReadableConfig flinkConf;
+ private final int threadNum;
private final JobConfWrapper jobConfWrapper;
private final List<String> partitionKeys;
private final ContinuousPartitionFetcher<Partition, ?> fetcher;
@@ -73,7 +72,7 @@ public class HiveSource<T> extends AbstractFileSource<T, HiveSourceSplit> {
FileSplitAssigner.Provider splitAssigner,
BulkFormat<T, HiveSourceSplit> readerFormat,
@Nullable ContinuousEnumerationSettings continuousEnumerationSettings,
- ReadableConfig flinkConf,
+ int threadNum,
JobConf jobConf,
ObjectPath tablePath,
List<String> partitionKeys,
@@ -86,10 +85,10 @@ public class HiveSource<T> extends AbstractFileSource<T, HiveSourceSplit> {
readerFormat,
continuousEnumerationSettings);
Preconditions.checkArgument(
- flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM) >= 1,
+ threadNum >= 1,
HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM.key()
+ " cannot be less than 1");
- this.flinkConf = flinkConf;
+ this.threadNum = threadNum;
this.jobConfWrapper = new JobConfWrapper(jobConf);
this.tablePath = tablePath;
this.partitionKeys = partitionKeys;
@@ -164,7 +163,7 @@ public class HiveSource<T> extends AbstractFileSource<T, HiveSourceSplit> {
seenPartitions,
getAssignerFactory().create(new ArrayList<>(splits)),
getContinuousEnumerationSettings().getDiscoveryInterval().toMillis(),
- flinkConf,
+ threadNum,
jobConfWrapper.conf(),
tablePath,
fetcher,
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java
index 25f7ed7..75b6e74 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java
@@ -68,6 +68,7 @@ import static org.apache.flink.connectors.hive.HiveOptions.STREAMING_SOURCE_ENAB
import static org.apache.flink.connectors.hive.HiveOptions.STREAMING_SOURCE_MONITOR_INTERVAL;
import static org.apache.flink.connectors.hive.HiveOptions.STREAMING_SOURCE_PARTITION_INCLUDE;
import static org.apache.flink.connectors.hive.HiveOptions.STREAMING_SOURCE_PARTITION_ORDER;
+import static org.apache.flink.connectors.hive.HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER;
import static org.apache.flink.table.catalog.hive.util.HiveTableUtil.checkAcidTable;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -78,7 +79,8 @@ public class HiveSourceBuilder {
private static final Duration DEFAULT_SCAN_MONITOR_INTERVAL = Duration.ofMinutes(1L);
private final JobConf jobConf;
- private final ReadableConfig flinkConf;
+ private final int threadNum;
+ private final boolean fallbackMappedReader;
private final ObjectPath tablePath;
private final Map<String, String> tableOptions;
@@ -110,7 +112,9 @@ public class HiveSourceBuilder {
@Nonnull String tableName,
@Nonnull Map<String, String> tableOptions) {
this.jobConf = jobConf;
- this.flinkConf = flinkConf;
+ this.threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ this.fallbackMappedReader = flinkConf.get(TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER);
this.tablePath = new ObjectPath(dbName, tableName);
this.hiveVersion = hiveVersion == null ? HiveShimLoader.getHiveVersion() : hiveVersion;
HiveConf hiveConf = HiveConfUtils.create(jobConf);
@@ -147,7 +151,9 @@ public class HiveSourceBuilder {
@Nullable String hiveVersion,
@Nonnull CatalogTable catalogTable) {
this.jobConf = jobConf;
- this.flinkConf = flinkConf;
+ this.threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ this.fallbackMappedReader = flinkConf.get(TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER);
this.tablePath = tablePath;
this.hiveVersion = hiveVersion == null ? HiveShimLoader.getHiveVersion() : hiveVersion;
this.fullSchema = catalogTable.getSchema();
@@ -231,12 +237,12 @@ public class HiveSourceBuilder {
new Path[1],
new HiveSourceFileEnumerator.Provider(
partitions != null ? partitions : Collections.emptyList(),
- flinkConf,
+ threadNum,
new JobConfWrapper(jobConf)),
splitAssigner,
bulkFormat,
continuousSourceSettings,
- flinkConf,
+ threadNum,
jobConf,
tablePath,
partitionKeys,
@@ -318,7 +324,7 @@ public class HiveSourceBuilder {
fullSchema.getFieldDataTypes(),
hiveVersion,
getProducedRowType(),
- flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER)),
+ fallbackMappedReader),
limit);
}
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceFileEnumerator.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceFileEnumerator.java
index 4227c36..dd0fe98 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceFileEnumerator.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceFileEnumerator.java
@@ -18,7 +18,6 @@
package org.apache.flink.connectors.hive;
-import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.enumerate.FileEnumerator;
import org.apache.flink.connectors.hive.read.HiveSourceSplit;
@@ -45,33 +44,27 @@ public class HiveSourceFileEnumerator implements FileEnumerator {
// For non-partition hive table, partitions only contains one partition which partitionValues is
// empty.
private final List<HiveTablePartition> partitions;
- private final ReadableConfig flinkConf;
+ private final int threadNum;
private final JobConf jobConf;
public HiveSourceFileEnumerator(
- List<HiveTablePartition> partitions, ReadableConfig flinkConf, JobConf jobConf) {
+ List<HiveTablePartition> partitions, int threadNum, JobConf jobConf) {
this.partitions = partitions;
- this.flinkConf = flinkConf;
+ this.threadNum = threadNum;
this.jobConf = jobConf;
}
@Override
public Collection<FileSourceSplit> enumerateSplits(Path[] paths, int minDesiredSplits)
throws IOException {
- return new ArrayList<>(createInputSplits(minDesiredSplits, partitions, flinkConf, jobConf));
+ return new ArrayList<>(createInputSplits(minDesiredSplits, partitions, threadNum, jobConf));
}
public static List<HiveSourceSplit> createInputSplits(
- int minNumSplits,
- List<HiveTablePartition> partitions,
- ReadableConfig flinkConf,
- JobConf jobConf)
+ int minNumSplits, List<HiveTablePartition> partitions, int threadNum, JobConf jobConf)
throws IOException {
List<HiveSourceSplit> hiveSplits = new ArrayList<>();
- try (MRSplitsGetter splitsGetter =
- new MRSplitsGetter(
- flinkConf.get(
- HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM))) {
+ try (MRSplitsGetter splitsGetter = new MRSplitsGetter(threadNum)) {
for (HiveTablePartitionSplits partitionSplits :
splitsGetter.getHiveTablePartitionMRSplits(minNumSplits, partitions, jobConf)) {
HiveTablePartition partition = partitionSplits.getHiveTablePartition();
@@ -109,21 +102,19 @@ public class HiveSourceFileEnumerator implements FileEnumerator {
private static final long serialVersionUID = 1L;
private final List<HiveTablePartition> partitions;
- private final ReadableConfig flinkConf;
+ private final int threadNum;
private final JobConfWrapper jobConfWrapper;
public Provider(
- List<HiveTablePartition> partitions,
- ReadableConfig flinkConf,
- JobConfWrapper jobConfWrapper) {
+ List<HiveTablePartition> partitions, int threadNum, JobConfWrapper jobConfWrapper) {
this.partitions = partitions;
- this.flinkConf = flinkConf;
+ this.threadNum = threadNum;
this.jobConfWrapper = jobConfWrapper;
}
@Override
public FileEnumerator create() {
- return new HiveSourceFileEnumerator(partitions, flinkConf, jobConfWrapper.conf());
+ return new HiveSourceFileEnumerator(partitions, threadNum, jobConfWrapper.conf());
}
}
}
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 710c890..4316670 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
@@ -108,7 +108,8 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
private static final Logger LOG = LoggerFactory.getLogger(HiveTableSink.class);
- private final ReadableConfig flinkConf;
+ private final boolean fallbackMappedReader;
+ private final boolean fallbackMappedWriter;
private final JobConf jobConf;
private final CatalogTable catalogTable;
private final ObjectIdentifier identifier;
@@ -128,7 +129,24 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
ObjectIdentifier identifier,
CatalogTable table,
@Nullable Integer configuredParallelism) {
- this.flinkConf = flinkConf;
+ this(
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER),
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_WRITER),
+ jobConf,
+ identifier,
+ table,
+ configuredParallelism);
+ }
+
+ private HiveTableSink(
+ boolean fallbackMappedReader,
+ boolean fallbackMappedWriter,
+ JobConf jobConf,
+ ObjectIdentifier identifier,
+ CatalogTable table,
+ @Nullable Integer configuredParallelism) {
+ this.fallbackMappedReader = fallbackMappedReader;
+ this.fallbackMappedWriter = fallbackMappedWriter;
this.jobConf = jobConf;
this.identifier = identifier;
this.catalogTable = table;
@@ -298,7 +316,7 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
org.apache.flink.core.fs.Path path = new org.apache.flink.core.fs.Path(sd.getLocation());
BucketsBuilder<RowData, String, ? extends BucketsBuilder<RowData, ?, ?>> builder;
- if (flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_WRITER)) {
+ if (fallbackMappedWriter) {
builder =
bucketsBuilderForMRWriter(
recordWriterFactory, sd, assigner, rollingPolicy, outputFileConfig);
@@ -377,7 +395,7 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
catalogTable,
hiveVersion,
(RowType) tableSchema.toRowDataType().getLogicalType(),
- flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER));
+ fallbackMappedReader);
}
private HiveTableMetaStoreFactory msFactory() {
@@ -487,7 +505,12 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
public DynamicTableSink copy() {
HiveTableSink sink =
new HiveTableSink(
- flinkConf, jobConf, identifier, catalogTable, configuredParallelism);
+ fallbackMappedReader,
+ fallbackMappedWriter,
+ jobConf,
+ identifier,
+ catalogTable,
+ configuredParallelism);
sink.staticPartitionSpec = staticPartitionSpec;
sink.overwrite = overwrite;
sink.dynamicGrouping = dynamicGrouping;
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
index d7ee578..cc5d639 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
@@ -146,6 +146,8 @@ public class HiveTableSource
catalogTable.getPartitionKeys(),
remainingPartitions);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
int parallelism =
new HiveParallelismInference(tablePath, flinkConf)
.infer(
@@ -156,7 +158,7 @@ public class HiveTableSource
HiveSourceFileEnumerator.createInputSplits(
0,
hivePartitionsToRead,
- flinkConf,
+ threadNum,
jobConf)
.size())
.limit(limit);
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormatPartitionReader.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormatPartitionReader.java
index 534c623..2493942 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormatPartitionReader.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormatPartitionReader.java
@@ -18,7 +18,6 @@
package org.apache.flink.connectors.hive.read;
-import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.file.table.PartitionReader;
import org.apache.flink.connectors.hive.HiveTablePartition;
import org.apache.flink.connectors.hive.JobConfWrapper;
@@ -36,7 +35,7 @@ public class HiveInputFormatPartitionReader
implements PartitionReader<HiveTablePartition, RowData> {
private static final long serialVersionUID = 1L;
- private final ReadableConfig flinkConf;
+ private final int threadNum;
private final JobConfWrapper jobConfWrapper;
private final String hiveVersion;
protected final ObjectPath tablePath;
@@ -51,7 +50,7 @@ public class HiveInputFormatPartitionReader
private transient int readingSplitId;
public HiveInputFormatPartitionReader(
- ReadableConfig flinkConf,
+ int threadNum,
JobConf jobConf,
String hiveVersion,
ObjectPath tablePath,
@@ -60,7 +59,7 @@ public class HiveInputFormatPartitionReader
List<String> partitionKeys,
int[] selectedFields,
boolean useMapRedReader) {
- this.flinkConf = flinkConf;
+ this.threadNum = threadNum;
this.jobConfWrapper = new JobConfWrapper(jobConf);
this.hiveVersion = hiveVersion;
this.tablePath = tablePath;
@@ -75,7 +74,7 @@ public class HiveInputFormatPartitionReader
public void open(List<HiveTablePartition> partitions) throws IOException {
hiveTableInputFormat =
new HiveTableInputFormat(
- this.flinkConf,
+ this.threadNum,
this.jobConfWrapper.conf(),
this.partitionKeys,
this.fieldTypes,
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveTableInputFormat.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveTableInputFormat.java
index 5297812..eadd114 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveTableInputFormat.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveTableInputFormat.java
@@ -23,8 +23,6 @@ import org.apache.flink.api.common.io.CheckpointableInputFormat;
import org.apache.flink.api.common.io.LocatableInputSplitAssigner;
import org.apache.flink.api.common.io.statistics.BaseStatistics;
import org.apache.flink.api.java.hadoop.common.HadoopInputFormatCommonBase;
-import org.apache.flink.configuration.ReadableConfig;
-import org.apache.flink.connectors.hive.HiveOptions;
import org.apache.flink.connectors.hive.HiveTablePartition;
import org.apache.flink.connectors.hive.HiveTablePartitionSplits;
import org.apache.flink.connectors.hive.JobConfWrapper;
@@ -69,7 +67,7 @@ public class HiveTableInputFormat extends HadoopInputFormatCommonBase<RowData, H
private static final String SCHEMA_EVOLUTION_COLUMNS = "schema.evolution.columns";
private static final String SCHEMA_EVOLUTION_COLUMNS_TYPES = "schema.evolution.columns.types";
- private final ReadableConfig flinkConf;
+ private final int threadNum;
private final JobConfWrapper jobConf;
@@ -98,7 +96,7 @@ public class HiveTableInputFormat extends HadoopInputFormatCommonBase<RowData, H
@VisibleForTesting protected transient SplitReader reader;
public HiveTableInputFormat(
- ReadableConfig flinkConf,
+ int threadNum,
JobConf jobConf,
List<String> partitionKeys,
DataType[] fieldTypes,
@@ -109,7 +107,7 @@ public class HiveTableInputFormat extends HadoopInputFormatCommonBase<RowData, H
boolean useMapRedReader,
List<HiveTablePartition> partitions) {
super(jobConf.getCredentials());
- this.flinkConf = flinkConf;
+ this.threadNum = threadNum;
this.jobConf = new JobConfWrapper(new JobConf(jobConf));
this.partitionKeys = partitionKeys;
this.fieldTypes = fieldTypes;
@@ -317,21 +315,15 @@ public class HiveTableInputFormat extends HadoopInputFormatCommonBase<RowData, H
@Override
public HiveTableInputSplit[] createInputSplits(int minNumSplits) throws IOException {
- return createInputSplits(minNumSplits, partitions, flinkConf, jobConf.conf());
+ return createInputSplits(minNumSplits, partitions, threadNum, jobConf.conf());
}
public static HiveTableInputSplit[] createInputSplits(
- int minNumSplits,
- List<HiveTablePartition> partitions,
- ReadableConfig flinkConf,
- JobConf jobConf)
+ int minNumSplits, List<HiveTablePartition> partitions, int threadNum, JobConf jobConf)
throws IOException {
List<HiveTableInputSplit> hiveSplits = new ArrayList<>();
int splitNum = 0;
- try (MRSplitsGetter splitsGetter =
- new MRSplitsGetter(
- flinkConf.get(
- HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM))) {
+ try (MRSplitsGetter splitsGetter = new MRSplitsGetter(threadNum)) {
for (HiveTablePartitionSplits partitionSplits :
splitsGetter.getHiveTablePartitionMRSplits(minNumSplits, partitions, jobConf)) {
for (InputSplit inputSplit : partitionSplits.getInputSplits()) {
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/PartitionMonitorTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/PartitionMonitorTest.java
index f366f10..0816092 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/PartitionMonitorTest.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/PartitionMonitorTest.java
@@ -182,7 +182,8 @@ public class PartitionMonitorTest {
0L,
seenPartitionsSinceOffset,
tablePath,
- configuration,
+ configuration.get(
+ HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM),
jobConf,
continuousPartitionFetcher,
fetcherContext);
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/read/HiveInputFormatPartitionReaderITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/read/HiveInputFormatPartitionReaderITCase.java
index bc8b971..aac9007 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/read/HiveInputFormatPartitionReaderITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/read/HiveInputFormatPartitionReaderITCase.java
@@ -17,7 +17,7 @@
package org.apache.flink.connectors.hive.read;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connectors.hive.HiveOptions;
import org.apache.flink.connectors.hive.HiveTablePartition;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.TableEnvironment;
@@ -62,7 +62,7 @@ public class HiveInputFormatPartitionReaderITCase {
// create partition reader
HiveInputFormatPartitionReader partitionReader =
new HiveInputFormatPartitionReader(
- new Configuration(),
+ HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM.defaultValue(),
new JobConf(hiveCatalog.getHiveConf()),
hiveCatalog.getHiveVersion(),
tablePath,