[FLINK-7244] Add parquet table source #8064

HuangZhenQiu · 2019-03-27T16:08:52Z

What is the purpose of the change

Add projectable and filterable ParquetTableSource based on ParquetInputFormat

Brief change log

Add Parquet table source with expression conversion to Parquet Filters
Add unit test and integration test to test the correctness of Fields Projection and Predicate Push

Verifying this change

Added unit test for public functions of ParquetTableSource
Added integration tests for end-to-end test ParquetTableSource in batch mode

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

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, 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? (not applicable)

flinkbot · 2019-03-27T16:10:32Z

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 b793cb3 (Fri Aug 23 10:13:21 UTC 2019)

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

fhueske

Hi @HuangZhenQiu,
Thanks for the PR!
I've added a first round of comments but didn't have a look at the tests yet.
Will do that in the next days.

Thanks, Fabian

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

fhueske · 2019-04-16T20:52:00Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ // the configuration to read the file
+ private final Configuration parquetConfig;
+
+ // type information of the data returned by the InputFormat


indent with tab

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

...s/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetTableSourceITCase.java

fhueske · 2019-04-16T21:10:05Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ private final RowTypeInfo typeInfo;
+ // list of selected Parquet fields to return
+ private final int[] selectedFields;
+ // list of predicates to apply


change to predicate expressions to apply since it's not really a list?

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

fhueske · 2019-04-16T21:49:02Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ }
+ }
+ } else if (exp instanceof BinaryExpression) {
+ FilterPredicate c1 = toParquetPredicate(((Or) exp).left());


exp was only checked to be a BinaryExpression so it could also be an And. However, we know that we receive the predicates in CNF, so there should be no Ands. I think we can check for Or instead of And before.

fhueske · 2019-04-16T21:49:23Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ } else {
+ if (exp instanceof Or) {
+ return FilterApi.or(c1, c2);
+ } else if (exp instanceof And) {


there should be no Ands because we get the predicates in CNF

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

HuangZhenQiu · 2019-04-23T14:15:59Z

@fhueske
I just resolved your first around of comments. Please add have another look when you have time.

fhueske · 2019-04-29T13:28:10Z

Sorry, I was out for a few days vacation. Will continue with my review soon.
Thanks, Fabian

fhueske

Thanks for your patience and the update @HuangZhenQiu!
I left a couple of comments.

Best, Fabian

flink-formats/flink-parquet/pom.xml

fhueske · 2019-06-11T20:39:42Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+
+ @Override
+ public boolean isFilterPushedDown() {
+ return predicate != null;


we can set a flag when applyPredicate() was called and return it here. This avoids multiple calls of applyPredicate() if no predicate can be successfully converted.

fhueske · 2019-06-11T20:52:29Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ } else if (exp instanceof BinaryExpression) {
+ if (exp instanceof And) {
+ LOG.debug("All of the predicates should be in CNF. Found an AND expression.", exp);
+ } else if (exp instanceof Or){


add space between ){

fhueske · 2019-06-11T20:54:03Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ if (c1 == null || c2 == null) {
+ return null;
+ } else {
+ if (exp instanceof Or) {


I think this condition can be removed. We already checked that exp instanceof Or

fhueske · 2019-06-11T20:54:40Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ } else {
+ if (exp instanceof Or) {
+ return FilterApi.or(c1, c2);
+ } else {


This else branch should be moved one condition outside.

fhueske · 2019-06-12T08:08:16Z

...ats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetTableSourceTest.java

+
+ // ensure table schema is the same
+ assertEquals(parquetTableSource.getTableSchema(), projected.getTableSchema());
+


Check that the source description is different (!parquetTableSource.explainSource.equals(projected.explainSource()))

fhueske · 2019-06-12T08:09:53Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+
+ @Override
+ public String explainSource() {
+ return "ParquetFile[path=" + path + ", schema=" + parquetSchema + ", filter=" + predicateString() + "]";


Include the projected fields in the source description. Flink checks whether the project pushdown worked by comparing the descriptions. If both are the same, the rule is not applied.

fhueske · 2019-06-12T08:11:39Z

...ats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetTableSourceTest.java

+
+ private ParquetTableSource createNestedTestParquetTableSource() throws Exception {
+ Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> nested = TestUtil.getNestedRecordTestData();
+ Path path = TestUtil.createTempParquetFile(tempRoot.getRoot(), TestUtil.NESTED_SCHEMA,


No need to create temp files here.

...ats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetTableSourceTest.java

fhueske · 2019-06-12T08:27:58Z

...ats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetTableSourceTest.java

+
+ // ensure table schema is identical
+ assertEquals(parquetTableSource.getTableSchema(), projected.getTableSchema());
+


Check that the pushed down predicates were removed from exps.

HuangZhenQiu · 2019-06-25T07:45:41Z

@fhueske I resolve all of the comments, except the last one. Not sure why "the pushed down predicates should be removed from exps.

fhueske · 2019-07-10T09:25:06Z

Hi @HuangZhenQiu, the predicates should be removed to let the optimizer know that these are handled by the table source. If they are not removed, the optimizer will generate code to evaluate the predicates which was already done by the table source.

I'll check the PR now.

fhueske

Thanks for the update @HuangZhenQiu!

The PR looks very good.
There are a few spots where we should respect the new code style guide, but I'll fix those and merge the PR.

Cheers, Fabian

fhueske · 2019-07-10T10:01:09Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ // type information of the data returned by the InputFormat
+ private final RowTypeInfo typeInfo;
+ // list of selected Parquet fields to return
+ private final int[] selectedFields;


Add @Nullable annotation as required by the new code style guidelines (apache/flink-web#224)

fhueske · 2019-07-10T10:01:15Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ // list of selected Parquet fields to return
+ private final int[] selectedFields;
+ // predicate expression to apply
+ private final FilterPredicate predicate;


Add @Nullable annotation

fhueske · 2019-07-10T10:01:40Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ }
+
+ private ParquetTableSource(String path, MessageType parquetSchema, Configuration configuration,
+ boolean recursiveEnumeration, int[] selectedFields, FilterPredicate predicate) {


Add @Nullable annotation to selectedFields and predicate

fhueske · 2019-07-10T10:02:00Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ /**
+ * Converts flink Expression to parquet FilterPredicate.
+ */
+ private FilterPredicate toParquetPredicate(Expression exp) {


Add @Nullable annotation

fhueske · 2019-07-10T12:32:51Z

...ats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetTableSourceTest.java

+ );
+
+ // ensure ParquetInputFormat is configured with selected fields
+ ParquetTableSource spyPTS = spy(projected);


our new code style guidelines discourage the use of Mockito.

fhueske · 2019-07-10T12:56:51Z

...formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

+ * </pre>
+ */
+public class ParquetTableSource
+ implements BatchTableSource<Row>, FilterableTableSource<Row>, ProjectableTableSource<Row> {


The new Blink query optimizer and execution is based on the InputFormatTableSource abstract class. We can simply implement both, BatchTableSource and InputFormatTableSource to support both of Flink's SQL engines.

OK, seems that we would need to add a dependency on flink-streaming for this. I'll keep it as it is for now.

This closes apache#8064.

flinkbot · 2019-07-10T16:23:40Z

CI report for commit b793cb3: SUCCESS Build

rmetzger added review=description? component=TableSQL/Ecosystem labels Mar 27, 2019

fhueske requested changes Apr 16, 2019

View reviewed changes

HuangZhenQiu force-pushed the add-parquet-table-source branch 3 times, most recently from 518ed2a to e49b8c7 Compare April 23, 2019 06:01

HuangZhenQiu force-pushed the add-parquet-table-source branch 6 times, most recently from e72ed37 to ff12d2d Compare June 1, 2019 19:46

fhueske requested changes Jun 12, 2019

View reviewed changes

HuangZhenQiu and others added 4 commits June 23, 2019 06:18

add unit test

ca56d7d

add integration test cases

1d8d2fa

resolve comments from fabian

0af091d

remove scala dependency

13befb3

HuangZhenQiu force-pushed the add-parquet-table-source branch 3 times, most recently from bad6b89 to e87a4d6 Compare June 23, 2019 22:47

resolve comments to reduce test file generation

b793cb3

HuangZhenQiu force-pushed the add-parquet-table-source branch from e87a4d6 to b793cb3 Compare June 24, 2019 01:34

fhueske approved these changes Jul 10, 2019

View reviewed changes

fhueske pushed a commit to fhueske/flink that referenced this pull request Jul 10, 2019

[FLINK-7244] Add ParquetTableSource.

ed8f35a

This closes apache#8064.

asfgit closed this in 38e5e81 Jul 10, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-7244] Add parquet table source #8064

[FLINK-7244] Add parquet table source #8064

HuangZhenQiu commented Mar 27, 2019

flinkbot commented Mar 27, 2019 •

edited

Loading

fhueske left a comment

fhueske Apr 16, 2019

fhueske Apr 16, 2019

fhueske Apr 16, 2019

fhueske Apr 16, 2019

HuangZhenQiu commented Apr 23, 2019

fhueske commented Apr 29, 2019

fhueske left a comment

fhueske Jun 11, 2019

fhueske Jun 11, 2019

fhueske Jun 11, 2019

fhueske Jun 11, 2019

fhueske Jun 12, 2019

fhueske Jun 12, 2019

fhueske Jun 12, 2019

fhueske Jun 12, 2019

HuangZhenQiu commented Jun 25, 2019

fhueske commented Jul 10, 2019

fhueske left a comment

fhueske Jul 10, 2019

fhueske Jul 10, 2019

fhueske Jul 10, 2019

fhueske Jul 10, 2019

fhueske Jul 10, 2019

fhueske Jul 10, 2019

fhueske Jul 10, 2019

flinkbot commented Jul 10, 2019


		// ensure table schema is the same
		assertEquals(parquetTableSource.getTableSchema(), projected.getTableSchema());


		// ensure table schema is identical
		assertEquals(parquetTableSource.getTableSchema(), projected.getTableSchema());

[FLINK-7244] Add parquet table source #8064

[FLINK-7244] Add parquet table source #8064

Conversation

HuangZhenQiu commented Mar 27, 2019

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 Mar 27, 2019 • edited Loading

Automated Checks

Review Progress

fhueske left a 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

HuangZhenQiu commented Apr 23, 2019

fhueske commented Apr 29, 2019

fhueske left a 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

HuangZhenQiu commented Jun 25, 2019

fhueske commented Jul 10, 2019

fhueske left a 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

flinkbot commented Jul 10, 2019

flinkbot commented Mar 27, 2019 •

edited

Loading