[FLINK-19365][hive] Migrate Hive source to FLIP-27 source interface f… #13915

lirui-apache · 2020-11-04T03:47:35Z

…or batch

What is the purpose of the change

Migrate hive connector to FLIP-27 source for batch reading

Brief change log

Implement HiveSourceSplit as sub-class of FileSourceSplit. It remembers the partition which a split belongs to.
Implement HiveSourceSplitSerializer as SerDe for HiveSourceSplit.
Implement HiveSource as sub-class of AbstractFileSource, and works with RowData and HiveSourceSplit.
Implement HiveSourceFileEnumerator as an implementation of FileEnumerator, but it generates splits based on partitions rather than paths.
Implement HiveBulkFormatAdapter as an implementation of BulkFormat<RowData, HiveSourceSplit>. It delegates the reading to other BulkFormat instances based on the information we have from each partition to read.
Update PartitionValueConverter to take partition col name when converting the partition value. This makes it easier for hive to reuse ParquetColumnarRowInputFormat because the partition values have been pre-computed.

Verifying this change

Existing tests

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): yes
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: yes
The runtime per-record code paths (performance sensitive): yes
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? NA

flinkbot · 2020-11-04T03:50:36Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 197131c (Wed Nov 04 03:50:36 UTC 2020)

Warnings:

1 pom.xml files were touched: Check for build and licensing issues.
No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-11-04T04:01:12Z

CI report:

a0ef581 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

...nnectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSource.java

JingsongLi · 2020-11-04T03:56:34Z

...nnectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSource.java

+			JobConf jobConf,
+			CatalogTable catalogTable,
+			List<HiveTablePartition> partitions,
+			int[] projectedFields,


Already has producedType, I think projectedFields can be removed

HiveMapredSplitReader still requires it. We can remove it when we refactor HiveMapredSplitReader

I think you can create projectedFields from producedType easily.

...nnectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSource.java

JingsongLi · 2020-11-04T03:59:15Z

...connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceSplitSerializer.java

+	}
+
+	private void serialize(ObjectOutputStream outputStream, HiveSourceSplit split) throws IOException {
+		byte[] superBytes = FileSourceSplitSerializer.INSTANCE.serialize(new FileSourceSplit(


Why do you want to create a new FileSourceSplit? Why not just use HiveSourceSplit

I think it's better to reuse FileSourceSplitSerializer to serialize the super class, but FileSourceSplitSerializer doesn't accept sub-classes.
Or we can just call InstantiationUtil.serializeObject to serialize the whole object. WDYT?

but FileSourceSplitSerializer doesn't accept sub-classes.

Oh, that is.

I am OK to keep this as it is.

...connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceSplitSerializer.java

...onnector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveBulkFormatAdapter.java

...link-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveSourceSplit.java

JingsongLi · 2020-11-04T05:50:03Z

...onnector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveBulkFormatAdapter.java

+					partitionKeys,
+					jobConfWrapper.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+							HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal),
+					(PartitionValueConverter) (colName, valStr, type) -> split.getHiveTablePartition().getPartitionSpec().get(colName),


This will fail when partition values can not be found in path.
I create #13919 to refactor this.

JingsongLi · 2020-11-04T05:57:07Z

...onnector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveBulkFormatAdapter.java

+			JobConf clonedConf = new JobConf(jobConfWrapper.conf());
+			addSchemaToConf(clonedConf);
+			HiveTableInputSplit oldSplit = new HiveTableInputSplit(-1, split.toMapRedSplit(), clonedConf, split.getHiveTablePartition());
+			hiveMapredSplitReader = new HiveMapredSplitReader(clonedConf, partitionKeys, fieldTypes, selectedFields, oldSplit, hiveShim);


If you don't want migrate HiveMapredSplitReader now, can you create JIRA for this?

Just created https://issues.apache.org/jira/browse/FLINK-19965 for that

…or batch

lirui-apache · 2020-11-04T12:10:13Z

@JingsongLi Thanks for the review. I have rebased and addressed your comments.

JingsongLi · 2020-11-05T02:11:21Z

...nnectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSource.java

+				new org.apache.flink.core.fs.Path[1],
+				new HiveSourceFileEnumerator.Provider(partitions, new JobConfWrapper(jobConf)),
+				DEFAULT_SPLIT_ASSIGNER,
+				createBulkFormat(new JobConf(jobConf), catalogTable, hiveVersion, producedDataType, useMapRedReader, limit),


I think you should use LimitableBulkFormat here.

JingsongLi · 2020-11-05T02:11:46Z

...-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveParallelismInference.java

-			parallelism = Math.min(parallelism, (int) limit / 1000);
+	int limit(Long limit) {
+		if (limit != null) {
+			parallelism = Math.min(parallelism, limit.intValue() / 1000);


Can you change limit.intValue() / 1000 to (int) (limit / 1000)

JingsongLi · 2020-11-05T02:12:23Z

...onnector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveBulkFormatAdapter.java

+	private final RowType producedDataType;
+	private final boolean useMapRedReader;
+	// We should limit the input read count of the splits, null represents no limit.
+	private final Long limit;


Remove limit things in this class

JingsongLi · 2020-11-05T02:12:46Z

...onnector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveBulkFormatAdapter.java

+	private final DataType[] fieldTypes;
+	private final String hiveVersion;
+	private final HiveShim hiveShim;
+	private final RowType producedDataType;


producedRowType

JingsongLi · 2020-11-05T02:13:03Z

...nnectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSource.java

+			String hiveVersion,
+			boolean useMapRedReader,
+			boolean isStreamingSource,
+			RowType producedDataType) {


producedRowType

JingsongLi · 2020-11-05T02:15:07Z

...onnector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveBulkFormatAdapter.java

+		}
+
+		private boolean reachLimit() {
+			return limit != null && numRead >= limit;


Please remove limit things

...ors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java

JingsongLi

Looks good to me, Thanks @lirui-apache for update.

rmetzger added the review=description? label Nov 4, 2020

JingsongLi requested changes Nov 4, 2020

View reviewed changes

rmetzger added the component=Connectors/Hive label Nov 4, 2020

JingsongLi reviewed Nov 4, 2020

View reviewed changes

lirui-apache added 3 commits November 4, 2020 19:18

[FLINK-19365][hive] Migrate Hive source to FLIP-27 source interface f…

6ed0a96

…or batch

address comments

fe892ec

address comments

54abdd1

lirui-apache force-pushed the FLINK-19365 branch from c85d007 to 54abdd1 Compare November 4, 2020 12:03

JingsongLi reviewed Nov 5, 2020

View reviewed changes

address comments

a0ef581

JingsongLi reviewed Nov 5, 2020

View reviewed changes

...ors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java Show resolved Hide resolved

JingsongLi approved these changes Nov 5, 2020

View reviewed changes

JingsongLi closed this Nov 5, 2020

JingsongLi reopened this Nov 5, 2020

JingsongLi merged commit 56207e8 into apache:master Nov 5, 2020

lirui-apache deleted the FLINK-19365 branch November 5, 2020 08:20

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-19365][hive] Migrate Hive source to FLIP-27 source interface f… #13915

[FLINK-19365][hive] Migrate Hive source to FLIP-27 source interface f… #13915

lirui-apache commented Nov 4, 2020

flinkbot commented Nov 4, 2020

flinkbot commented Nov 4, 2020 •

edited

JingsongLi Nov 4, 2020

lirui-apache Nov 4, 2020

JingsongLi Nov 4, 2020

JingsongLi Nov 4, 2020

lirui-apache Nov 4, 2020

JingsongLi Nov 4, 2020

JingsongLi Nov 4, 2020

JingsongLi Nov 4, 2020

JingsongLi Nov 4, 2020

lirui-apache Nov 4, 2020

lirui-apache commented Nov 4, 2020

JingsongLi Nov 5, 2020

JingsongLi Nov 5, 2020

JingsongLi Nov 5, 2020

JingsongLi Nov 5, 2020

JingsongLi Nov 5, 2020

JingsongLi Nov 5, 2020

JingsongLi left a comment

[FLINK-19365][hive] Migrate Hive source to FLIP-27 source interface f… #13915

[FLINK-19365][hive] Migrate Hive source to FLIP-27 source interface f… #13915

Conversation

lirui-apache commented Nov 4, 2020

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Nov 4, 2020

Automated Checks

Review Progress

flinkbot commented Nov 4, 2020 • edited

CI report:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lirui-apache commented Nov 4, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JingsongLi left a comment

Choose a reason for hiding this comment

flinkbot commented Nov 4, 2020 •

edited