Added Filter and Projection Pushdown support to ParquetIO and Beam SQL's ParquetTable #35589

talatuyarer · 2025-07-15T05:15:23Z

This PR fixes #19748

This pull request introduces support for filter and projection pushdown to the ParquetIO connector when used as a Beam SQL source, significantly improving query performance by reducing the amount of data read from disk and transferred over the network.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

github-actions · 2025-07-15T05:41:40Z

Assigning reviewers:

R: @m-trieu for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

ahmedabu98

Thanks @talatuyarer, left some comments

ahmedabu98 · 2025-07-15T18:05:53Z

sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetFilterFactory.java

+  private static FilterPredicate convert(RexNode e, Schema beamSchema) {
+    SqlKind kind = e.getKind();


Do we need to validate that RexNode e is indeed a RexCall?

ahmedabu98 · 2025-07-15T18:09:34Z

sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetFilterFactory.java

+    List<RexNode> valueNodes = call.getOperands().subList(1, call.getOperands().size());
+    SqlKind comparison = isNotIn ? SqlKind.NOT_EQUALS : SqlKind.EQUALS;
+
+    // CHANGE: Use an explicit loop


Elaborate what this means? Should this be a TODO?

ahmedabu98 · 2025-07-15T18:16:32Z

sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetFilterFactory.java

+    return beamSchema.getField(columnRef.getIndex()).getName();
+  }
+
+  private static FilterPredicate createIntPredicate(SqlKind kind, String name, Integer value) {


I think the create___Predicate methods can be merged into one method. A lot of it seems to be duplicated logic

ahmedabu98 · 2025-07-15T18:19:59Z

sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetFilterFactory.java

+      case DATE:
+      case TIME:
+      case TIMESTAMP:
+      case ARRAY:
+      case MAP:
+      case ROW:
+      default:
+        return null;


This will suppress bad filters and may confuse users who expect it to work. Can we throw an UnsupportedOperationException instead?

ahmedabu98 · 2025-07-15T18:25:51Z

sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetFilterFactory.java

+      default:
+        return null;


throw an UnsupportedOperationException?

ahmedabu98 · 2025-07-15T19:31:28Z

sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetIO.java

@@ -340,6 +345,12 @@ public Read withProjection(Schema projectionSchema, Schema encoderSchema) {
          .build();
    }

+    /** Specifies a filter predicate to use for filtering records. */
+    public Read withFilter(ParquetFilter predicate) {


How would a Java user make use of this? This sink relies on ParquetFilter being an instance of ParquetFilterImpl, which is not available to the user.

ahmedabu98 · 2025-07-15T19:34:22Z

sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetFilterFactory.java

+import org.apache.parquet.filter2.predicate.Operators.LongColumn;
+import org.apache.parquet.io.api.Binary;
+
+public class ParquetFilterFactory {


We need thorough testing for this class

ahmedabu98 · 2025-07-15T19:50:42Z

...sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTable.java

-      Schema projectionSchema = projectSchema(schema, fieldNames);
-      LOG.info("Projecting fields schema: {}", projectionSchema);
-      read = read.withProjection(projectionSchema, projectionSchema);
+    read = read.withProjection(readSchema, readSchema);


I think we should be making the unwanted columns nullable for the second readSchema. Check ParquetIO javadoc:

beam/sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetIO.java

Lines 137 to 141 in cd72a25

* <p>Reading with projection can be enabled with the projection schema as following. Splittable

* reading is enabled when reading with projection. The projection_schema contains only the column

* that we would like to read and encoder_schema contains the schema to encode the output with the

* unwanted columns changed to nullable. Partial reading provide decrease of reading time due to

* partial processing of the data and partial encoding. The decrease in the reading time depends on

ahmedabu98 · 2025-07-15T19:55:32Z

...he/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTableProviderFilterPushDownTest.java

+  @Parameter(3)
+  public long expectedReadCount;
+
+  @Parameter(4)
+  public List<Row> expectedRows;
+
+  @Parameter(5)
+  public Schema expectedSchema;


I think we don't need expectedReadCount or expectedSchema, right? That information is already captured in expectedRows

ahmedabu98 · 2025-07-15T19:58:13Z

...he/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTableProviderFilterPushDownTest.java

+            "SELECT * FROM ProductInfo WHERE price > ? AND is_stocked = ?",
+            Arrays.asList(100.0, true),


Are the values inside the array supposed to be substituted for "?" ? If so, the output seems to be incorrect

talatuyarer added 3 commits July 14, 2025 22:11

Initial SQL ParquetTable predicate support

5a4e8fc

Initial SQL ParquetTable predicate support

5826071

Resolved merge conflict

fec97c3

github-actions bot added java io extensions sql parquet labels Jul 15, 2025

github-actions bot added the Next Action: Reviewers label Jul 15, 2025

ahmedabu98 reviewed Jul 15, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Added Filter and Projection Pushdown support to ParquetIO and Beam SQL's ParquetTable #35589

Added Filter and Projection Pushdown support to ParquetIO and Beam SQL's ParquetTable #35589

talatuyarer commented Jul 15, 2025

Uh oh!

github-actions bot commented Jul 15, 2025

Uh oh!

ahmedabu98 left a comment

Uh oh!

ahmedabu98 Jul 15, 2025

Uh oh!

ahmedabu98 Jul 15, 2025

Uh oh!

ahmedabu98 Jul 15, 2025

Uh oh!

ahmedabu98 Jul 15, 2025

Uh oh!

ahmedabu98 Jul 15, 2025

Uh oh!

ahmedabu98 Jul 15, 2025

Uh oh!

ahmedabu98 Jul 15, 2025

Uh oh!

ahmedabu98 Jul 15, 2025

Uh oh!

ahmedabu98 Jul 15, 2025

Uh oh!

ahmedabu98 Jul 15, 2025

Uh oh!

Uh oh!

		private static FilterPredicate convert(RexNode e, Schema beamSchema) {
		SqlKind kind = e.getKind();

	* <p>Reading with projection can be enabled with the projection schema as following. Splittable
	* reading is enabled when reading with projection. The projection_schema contains only the column
	* that we would like to read and encoder_schema contains the schema to encode the output with the
	* unwanted columns changed to nullable. Partial reading provide decrease of reading time due to
	* partial processing of the data and partial encoding. The decrease in the reading time depends on

		"SELECT * FROM ProductInfo WHERE price > ? AND is_stocked = ?",
		Arrays.asList(100.0, true),

Added Filter and Projection Pushdown support to ParquetIO and Beam SQL's ParquetTable #35589

Are you sure you want to change the base?

Added Filter and Projection Pushdown support to ParquetIO and Beam SQL's ParquetTable #35589

Conversation

talatuyarer commented Jul 15, 2025

GitHub Actions Tests Status (on master branch)

Uh oh!

github-actions bot commented Jul 15, 2025

Uh oh!

ahmedabu98 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!