Support Aggregate push down for incremental scan #7636

huaxingao · 2023-05-17T21:35:41Z

Enable Aggregate push down for incremental scan

huaxingao · 2023-05-18T00:41:39Z

huaxingao · 2023-05-18T00:42:31Z

also cc @amogh-jahagirdar @singhpk234 @nastra @ChristinaTech

RussellSpitzer · 2023-05-18T16:00:56Z

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java

-    if (snapshot == null) {
-      LOG.info("Skipping aggregate pushdown: table snapshot is null");
-      return false;
+    org.apache.iceberg.Scan scan;


minor style thought?

Could we do something like

Scan scan = null; if (readConf.startSnapshotId() == null) { TableScan tableScan = table.newScan() Snapshot snapshot = readSnapshot(); if (snapshot == null) { LOG.info("Skipping aggregate pushdown: table snapshot is null"); return false; } tableScan = tableScan.useSnapshot(snapshot.snapshotId()); scan = tableScan } else { IncrementalAppendScan incrementalScan = table.newIncrementalAppendScan(); incrementalScan = incrementalScan.fromSnapshotExclusive(readConf.startSnapshotId()); Long endSnapshotId = readConf.endSnapshotId(); if (endSnapshotId != null) { incrementalScan = incrementalScan.toSnapshot(endSnapshotId); } scan = incrementalScan; } scan = scan.filter(filterExpression()).includeColumnStats();

I could also see extracting that branching part into another function. My main suggestion is basically to structure it more like

if (tableScan) { TableScanStuff } else () { IncrementalStuff } Common Stuff

or

def Scan getScan() { ... } getScan() .filter() .includeColumnStats()

So we can remove all the casting and keep all the common code in one place just incase we need to add things in the future

Look further, we already have a build scan procedure for normal scans, why don't we use those private functions? or extract that logic?

Like buildBatchScan

Thank you very much for your comment! I have changed the code to reuse the buildBatchScan logic. Could you please take a look again when you have a moment?

ChristinaTech · 2023-05-19T18:22:24Z

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java

-    scan = scan.useSnapshot(snapshot.snapshotId());
-    scan = configureSplitPlanning(scan);
-    scan = scan.filter(filterExpression());
+    org.apache.iceberg.Scan scan = buildIcebergBatchScan(true);


Please modify boolean parameters you are passing with an inline comment to follow the Contributing Style Guidelines on Boolean Arguments

Fixed. Thanks

ChristinaTech · 2023-05-19T18:36:48Z

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java

@@ -188,6 +187,7 @@ public Filter[] pushedFilters() {
  }

  @Override
+  @SuppressWarnings("checkstyle:CyclomaticComplexity")


You have removed a decent amount of the branch points that were added in the prior commit that made this SuppressWarnings necessary. Can you see if its still needed after the most recent refactoring?

Removed. Thanks!

RussellSpitzer · 2023-05-19T19:16:26Z

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java

+        spark, table, buildIcebergBatchScan(false), readConf, expectedSchema, filterExpressions);
+  }
+
+  private org.apache.iceberg.Scan buildIcebergBatchScan(boolean withStats) {


nit: Do we need the fully qualified type?

Yea, we need the fully qualified type because we have already import org.apache.spark.sql.connector.read.Scan

RussellSpitzer · 2023-05-19T19:21:50Z

spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java

-            .collectAsList();
-    Assert.assertEquals("Records should match", expectedRecords.subList(1, 4), result);
+            .load(tableLocation);
+    List<SimpleRecord> result1 =


Any way we can check that pushdown is being used? I know this test passes now but I just want to make sure we are also using pushdown in future test scenarios

Yes, we can check if pushdown is being used. I normally check the explain string to find out if pushdown is being used.
For example,

SELECT min(data), max(data), count(data) FROM table;

If aggregate is pushed down, the physical plan has

LocalTableScan [min(data)#4461, max(data)#4462, count(data)#4463L]

If aggregate is not pushed down, the physical plan has

BatchScan default.table[data#4471]

I didn't check the explain string in this test to see if aggregate is pushed down, but I added a test for incremental Scan in TestAggregatePushDown. I have checked the explain string in that test to make sure aggregate is pushed down.

RussellSpitzer · 2023-05-19T19:23:18Z

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java

@@ -432,6 +426,10 @@ private Scan buildBatchScan(Long snapshotId, Long asOfTimestamp, String branch,
            .filter(filterExpression())
            .project(expectedSchema);

+    if (withStats) {
+      scan = scan.includeColumnStats();


Unrelated to this PR, but something we should think about is how this can be overly expensive on extremely wide tables. @pvary is actually dealing with a similar issue, we may want to extract just the columns relating to our aggregation to avoid memory issues. Not an issue for this PR though, just wanted to bring it up.

RussellSpitzer

This looks much cleaner! Thanks so much @huaxingao . I think once the remaining nits are cleaned up this is good to go. Just lets make sure we have a test where both aggregate pushdown and non-aggregate pushdown aggregates work successfully in Incr and Batch.

huaxingao · 2023-05-23T19:35:19Z

@RussellSpitzer I have addressed the comments and added a a new test in TestAggregatePushDown. Could you please take one more look when you have time? Thanks a lot!

RussellSpitzer · 2023-05-23T19:51:49Z

spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java

+            .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+    String explain2 =
+        dfWithoutAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+    explainContainsPushDownAggregates1 = false;


We have a StringAssert.contains(strings*) you can use (which will also give a nicer exception)

Assertions.asserThat("str").contains(.....)

Also how do we make sure that the function expressions are in the TableRelation and not somewhere else in the plan?

I don't have a good way to make sure that the expressions are in the TableRelation.

Here is the pushed down plan:

== Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- HashAggregate(keys=[], functions=[min(agg_func_0#51), max(agg_func_1#52), sum(agg_func_2#53L)]) +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [plan_id=65] +- HashAggregate(keys=[], functions=[partial_min(agg_func_0#51), partial_max(agg_func_1#52), partial_sum(agg_func_2#53L)]) +- Project [min(data)#54 AS agg_func_0#51, max(data)#55 AS agg_func_1#52, count(data)#56L AS agg_func_2#53L] +- LocalTableScan [min(data)#54, max(data)#55, count(data)#56L]

Here is the non pushed down plan:

== Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- HashAggregate(keys=[], functions=[min(data#414), max(data#414), count(data#414)]) +- HashAggregate(keys=[], functions=[partial_min(data#414), partial_max(data#414), partial_count(data#414)]) +- BatchScan spark_catalog.default.table[data#414] spark_catalog.default.table (branch=null) [filters=, groupedBy=] RuntimeFilters: []

I think actually it might be easier to check LocalTableScan instead.

I have changed this test to check LocalTableScan. I will have a followup to change all the other tests to also check LocalTableScan .

The assertions at 588. 567. and 611 all can be converted to assertJ style contains with multiple strings. Other than that I think this is probably fine for now.

RussellSpitzer

There are still a few assertJ conversions left to do in the tests but I think this is basically good to go. If I thought the underlying api would be more stable I would also push for testing the actual plans rather than the explain strings but I think they are both probably equally brittle in this case in terms of Spark upgrades.

I'll be on board to merge once the remaining tests are cleaned up

RussellSpitzer · 2023-05-24T15:47:29Z

spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java

@@ -288,29 +289,37 @@ public void testIncrementalScanOptions() throws IOException {
        });

    // test (1st snapshot, current snapshot] incremental scan.
-    List<SimpleRecord> result =
+    Dataset<Row> resultDf1 =


nit: resultDF1 => currentSnapshotResult

RussellSpitzer · 2023-05-24T15:47:54Z

spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java


    // test (2nd snapshot, 3rd snapshot] incremental scan.
-    Dataset<Row> resultDf =
+    Dataset<Row> resultDf2 =


resultDf2 => incrementalResult

RussellSpitzer · 2023-05-24T15:48:44Z

spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java

+    long snapshotId3 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+    sql("INSERT INTO %s VALUES (8, 7777), (9, 9999)", tableName);
+
+    Dataset<Row> dfWithAggPushdown1 =


pushdownResult

RussellSpitzer · 2023-05-24T15:48:53Z

spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java

+
+    Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates1);
+
+    Dataset<Row> dfWithoutAggPushdown1 =


noPushdownResult

RussellSpitzer · 2023-05-24T15:49:13Z

spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java

+        rowsToJava(dfWithAggPushdown1.collectAsList()),
+        rowsToJava(dfWithoutAggPushdown1.collectAsList()));
+
+    Dataset<Row> dfWithAggPushdown2 =


unboundedPushdownResult

RussellSpitzer · 2023-05-24T15:49:30Z

spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java

+
+    Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates2);
+
+    Dataset<Row> dfWithoutAggPushdown2 =


unboundedNoPushdownResult

RussellSpitzer · 2023-05-24T15:50:03Z

spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java

@@ -535,6 +540,106 @@ public void testAggregatePushDownForTimeTravel() {
    assertEquals("count push down", expected2, actual2);
  }

+  @Test
+  @SuppressWarnings("checkstyle:CyclomaticComplexity")


Added some suggested renames, but would also be good to just split the use cases into separate tests to avoid having complicated variable names

RussellSpitzer · 2023-05-24T15:50:47Z

spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java

@@ -285,29 +286,37 @@ public void testIncrementalScanOptions() throws IOException {
            "Cannot set only end-snapshot-id for incremental scans. Please, set start-snapshot-id too.");

    // test (1st snapshot, current snapshot] incremental scan.
-    List<SimpleRecord> result =
+    Dataset<Row> resultDf1 =


batchResult

RussellSpitzer · 2023-05-24T15:51:17Z

spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java


    // test (2nd snapshot, 3rd snapshot] incremental scan.
-    Dataset<Row> resultDf =
+    Dataset<Row> resultDf2 =


incrementalResult (again could just be two test cases, not picky about that or naming here but we should choose one)

ChristinaTech

Approved, though there are merge conflicts so this PR will likely need to be rebased.

szehon-ho · 2023-07-13T23:44:00Z

@RussellSpitzer was this close, and do you think it should go to 1.3.1?

Edit: realize its already disabled, so i guess it should not go into 1.3.1, as its a new feature and not a bug fix.

ChristinaTech · 2023-08-08T15:42:04Z

Branch has conflicts again and can't be merged. Please fix when you have the chance. Would be nice if we could get this into 1.4.

RussellSpitzer · 2023-08-08T16:10:03Z

The conflict is very small so i think we can just merge after we get that fixed

huaxingao · 2024-06-19T17:15:59Z

Sorry, I somehow missed the conversation thread for this PR and never resolved the conflicts. I noticed this when I recently checked my Iceberg PRs. It's a bit easier to open a new PR than to resolve the conflicts in this one. Here is the new PR.

I will add support in Spark 3.5 first, and then add back the changes in 3.4 and 3.3 afterwards.

I will close this PR for now.

github-actions bot added the spark label May 17, 2023

RussellSpitzer reviewed May 18, 2023

View reviewed changes

ChristinaTech suggested changes May 19, 2023

View reviewed changes

RussellSpitzer reviewed May 19, 2023

View reviewed changes

RussellSpitzer approved these changes May 19, 2023

View reviewed changes

ChristinaTech approved these changes May 20, 2023

View reviewed changes

RussellSpitzer reviewed May 23, 2023

View reviewed changes

RussellSpitzer approved these changes May 24, 2023

View reviewed changes

RussellSpitzer reviewed May 24, 2023

View reviewed changes

ChristinaTech approved these changes Jul 4, 2023

View reviewed changes

huaxingao added 7 commits July 13, 2023 14:08

Support Aggregate push down for incremental scan

a450f3f

fix 3.3

1e58650

Suppress CyclomaticComplexity

d1f2838

Suppress CyclomaticComplexity for 3.4

611bf6b

reuse build scan code

a136150

spotlessApply

1cafc75

address comments

07eeca2

huaxingao added 6 commits July 13, 2023 14:08

change this test to use StringAssert.contains

1780d0b

uses more descriptive names

0232d9f

formatting

28af35a

import Assertions

3bac3f8

fix error

157a67b

resolve conflicts

3c362a8

huaxingao force-pushed the agg_pushdown_incremental branch from d13be87 to 3c362a8 Compare July 13, 2023 21:14

huaxingao closed this Jun 19, 2024

huaxingao mentioned this pull request Jun 19, 2024

Support Aggregate push down for incremental scan #10538

Merged


		Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates1);

		Dataset<Row> dfWithoutAggPushdown1 =


		Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates2);

		Dataset<Row> dfWithoutAggPushdown2 =

Support Aggregate push down for incremental scan #7636

Support Aggregate push down for incremental scan #7636

Conversation

huaxingao commented May 17, 2023 • edited Loading

huaxingao commented May 18, 2023

huaxingao commented May 18, 2023

RussellSpitzer May 18, 2023 • edited Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RussellSpitzer left a comment

Choose a reason for hiding this comment

huaxingao commented May 23, 2023

RussellSpitzer May 23, 2023 • edited Loading

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

RussellSpitzer 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

Choose a reason for hiding this comment

ChristinaTech left a comment

Choose a reason for hiding this comment

szehon-ho commented Jul 13, 2023 • edited Loading

ChristinaTech commented Aug 8, 2023

RussellSpitzer commented Aug 8, 2023

huaxingao commented Jun 19, 2024 • edited Loading

huaxingao commented May 17, 2023 •

edited

Loading

RussellSpitzer May 18, 2023 •

edited

Loading

RussellSpitzer May 23, 2023 •

edited

Loading

szehon-ho commented Jul 13, 2023 •

edited

Loading

huaxingao commented Jun 19, 2024 •

edited

Loading