You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "huaxingao (via GitHub)" <gi...@apache.org> on 2023/05/17 21:35:43 UTC
[GitHub] [iceberg] huaxingao opened a new pull request, #7636: Support Aggregate push down for incremental scan
huaxingao opened a new pull request, #7636:
URL: https://github.com/apache/iceberg/pull/7636
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1204427426
##########
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 =
spark
.read()
.format("iceberg")
.option("start-snapshot-id", snapshotIds.get(3).toString())
- .load(tableLocation)
- .orderBy("id")
- .as(Encoders.bean(SimpleRecord.class))
- .collectAsList();
- Assert.assertEquals("Records should match", expectedRecords.subList(1, 4), result);
+ .load(tableLocation);
+ List<SimpleRecord> result1 =
+ resultDf1.orderBy("id").as(Encoders.bean(SimpleRecord.class)).collectAsList();
+ Assert.assertEquals("Records should match", expectedRecords.subList(1, 4), result1);
+ Assert.assertEquals("Unprocessed count should match record count", 3, resultDf1.count());
+
+ Row row1 = resultDf1.agg(functions.min("id"), functions.max("id")).head();
+ Assert.assertEquals("min value should match", 2, row1.getInt(0));
+ Assert.assertEquals("max value should match", 4, row1.getInt(1));
// test (2nd snapshot, 3rd snapshot] incremental scan.
- Dataset<Row> resultDf =
+ Dataset<Row> resultDf2 =
Review Comment:
incrementalResult (again could just be two test cases, not picky about that or naming here but we should choose one)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1204425085
##########
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")
+ public void testAggregatePushDownForIncrementalScan() {
+ sql("CREATE TABLE %s (id LONG, data INT) USING iceberg", tableName);
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, 1111), (1, 2222), (2, 3333), (2, 4444), (3, 5555), (3, 6666) ",
+ tableName);
+ long snapshotId1 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (4, 7777), (5, 8888)", tableName);
+ long snapshotId2 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (6, -7777), (7, 8888)", tableName);
+ long snapshotId3 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (8, 7777), (9, 9999)", tableName);
+
+ Dataset<Row> dfWithAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain1 =
+ dfWithAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ boolean explainContainsPushDownAggregates1 = false;
+ if (explain1.contains("count(data)")
+ && explain1.contains("min(data)")
+ && explain1.contains("max(data)")) {
+ explainContainsPushDownAggregates1 = true;
+ }
+
+ Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates1);
+
+ Dataset<Row> dfWithoutAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .option(SparkReadOptions.AGGREGATE_PUSH_DOWN_ENABLED, "false")
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain2 =
+ dfWithoutAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ explainContainsPushDownAggregates1 = false;
+ if (explain2.contains("count(data)")
+ || explain2.contains("min(data)")
+ || explain2.contains("max(data)")) {
+ explainContainsPushDownAggregates1 = true;
+ }
+
+ Assert.assertFalse("aggregate pushed down", explainContainsPushDownAggregates1);
+ assertEquals(
+ "Aggregate pushdown and non-aggregate pushdown should have the same results",
+ rowsToJava(dfWithAggPushdown1.collectAsList()),
+ rowsToJava(dfWithoutAggPushdown1.collectAsList()));
+
+ Dataset<Row> dfWithAggPushdown2 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId1)
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+
+ String explain3 =
+ dfWithAggPushdown2.queryExecution().explainString(ExplainMode.fromString("simple"));
+ boolean explainContainsPushDownAggregates2 = false;
+ if (explain3.contains("count(data)")
+ && explain3.contains("min(data)")
+ && explain3.contains("max(data)")) {
+ explainContainsPushDownAggregates2 = true;
+ }
+
+ Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates2);
+
+ Dataset<Row> dfWithoutAggPushdown2 =
Review Comment:
unboundedNoPushdownResult
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1199297419
##########
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 =
spark
.read()
.format("iceberg")
.option("start-snapshot-id", snapshotIds.get(3).toString())
- .load(tableLocation)
- .orderBy("id")
- .as(Encoders.bean(SimpleRecord.class))
- .collectAsList();
- Assert.assertEquals("Records should match", expectedRecords.subList(1, 4), result);
+ .load(tableLocation);
+ List<SimpleRecord> result1 =
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1202919345
##########
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")
+ public void testAggregatePushDownForIncrementalScan() {
+ sql("CREATE TABLE %s (id LONG, data INT) USING iceberg", tableName);
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, 1111), (1, 2222), (2, 3333), (2, 4444), (3, 5555), (3, 6666) ",
+ tableName);
+ long snapshotId1 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (4, 7777), (5, 8888)", tableName);
+ long snapshotId2 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (6, -7777), (7, 8888)", tableName);
+ long snapshotId3 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (8, 7777), (9, 9999)", tableName);
+
+ Dataset<Row> dfWithAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain1 =
+ dfWithAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ boolean explainContainsPushDownAggregates1 = false;
+ if (explain1.contains("count(data)")
+ && explain1.contains("min(data)")
+ && explain1.contains("max(data)")) {
+ explainContainsPushDownAggregates1 = true;
+ }
+
+ Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates1);
+
+ Dataset<Row> dfWithoutAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .option(SparkReadOptions.AGGREGATE_PUSH_DOWN_ENABLED, "false")
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain2 =
+ dfWithoutAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ explainContainsPushDownAggregates1 = false;
Review Comment:
We have a StringAssert.contains(strings*) you can use (which will also give a nicer exception)
Assertions.asserThat("str").contains(.....)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1198001598
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -221,15 +222,26 @@ public boolean pushAggregation(Aggregation aggregation) {
return false;
}
- TableScan scan = table.newScan().includeColumnStats();
- Snapshot snapshot = readSnapshot();
- if (snapshot == null) {
- LOG.info("Skipping aggregate pushdown: table snapshot is null");
- return false;
+ org.apache.iceberg.Scan scan;
Review Comment:
minor style thought?
Could we do something like
```java
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
```java
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1199558506
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -221,15 +221,7 @@ public boolean pushAggregation(Aggregation aggregation) {
return false;
}
- TableScan scan = table.newScan().includeColumnStats();
- Snapshot snapshot = readSnapshot();
- if (snapshot == null) {
- LOG.info("Skipping aggregate pushdown: table snapshot is null");
- return false;
- }
- scan = scan.useSnapshot(snapshot.snapshotId());
- scan = configureSplitPlanning(scan);
- scan = scan.filter(filterExpression());
+ org.apache.iceberg.Scan scan = buildIcebergBatchScan(true);
Review Comment:
Fixed. Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1204422541
##########
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 =
Review Comment:
nit: resultDF1 => currentSnapshotResult
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1199558522
##########
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")
Review Comment:
Removed. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1199145633
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -221,15 +222,26 @@ public boolean pushAggregation(Aggregation aggregation) {
return false;
}
- TableScan scan = table.newScan().includeColumnStats();
- Snapshot snapshot = readSnapshot();
- if (snapshot == null) {
- LOG.info("Skipping aggregate pushdown: table snapshot is null");
- return false;
+ org.apache.iceberg.Scan scan;
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1204423020
##########
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 =
spark
.read()
.format("iceberg")
.option("start-snapshot-id", snapshotIds.get(3).toString())
- .load(tableLocation)
- .orderBy("id")
- .as(Encoders.bean(SimpleRecord.class))
- .collectAsList();
- Assert.assertEquals("Records should match", expectedRecords.subList(1, 4), result);
+ .load(tableLocation);
+ List<SimpleRecord> result1 =
+ resultDf1.orderBy("id").as(Encoders.bean(SimpleRecord.class)).collectAsList();
+ Assert.assertEquals("Records should match", expectedRecords.subList(1, 4), result1);
+ Assert.assertEquals("Unprocessed count should match record count", 3, resultDf1.count());
+
+ Row row1 = resultDf1.agg(functions.min("id"), functions.max("id")).head();
+ Assert.assertEquals("min value should match", 2, row1.getInt(0));
+ Assert.assertEquals("max value should match", 4, row1.getInt(1));
// test (2nd snapshot, 3rd snapshot] incremental scan.
- Dataset<Row> resultDf =
+ Dataset<Row> resultDf2 =
Review Comment:
resultDf2 => incrementalResult
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1198006855
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -221,15 +222,26 @@ public boolean pushAggregation(Aggregation aggregation) {
return false;
}
- TableScan scan = table.newScan().includeColumnStats();
- Snapshot snapshot = readSnapshot();
- if (snapshot == null) {
- LOG.info("Skipping aggregate pushdown: table snapshot is null");
- return false;
+ org.apache.iceberg.Scan scan;
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1199558579
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -376,6 +363,12 @@ public Scan build() {
}
private Scan buildBatchScan() {
+ Schema expectedSchema = schemaWithMetadataColumns();
+ return new SparkBatchQueryScan(
+ spark, table, buildIcebergBatchScan(false), readConf, expectedSchema, filterExpressions);
+ }
+
+ private org.apache.iceberg.Scan buildIcebergBatchScan(boolean withStats) {
Review Comment:
Yea, we need the fully qualified type because we have already `import org.apache.spark.sql.connector.read.Scan`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1199298553
##########
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();
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on pull request #7636: Support Aggregate push down for incremental scan
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#issuecomment-1560018806
@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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] ChristinaTech commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "ChristinaTech (via GitHub)" <gi...@apache.org>.
ChristinaTech commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1199251036
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -221,15 +221,7 @@ public boolean pushAggregation(Aggregation aggregation) {
return false;
}
- TableScan scan = table.newScan().includeColumnStats();
- Snapshot snapshot = readSnapshot();
- if (snapshot == null) {
- LOG.info("Skipping aggregate pushdown: table snapshot is null");
- return false;
- }
- scan = scan.useSnapshot(snapshot.snapshotId());
- scan = configureSplitPlanning(scan);
- scan = scan.filter(filterExpression());
+ org.apache.iceberg.Scan scan = buildIcebergBatchScan(true);
Review Comment:
Please modify boolean parameters you are passing with an inline comment to follow the [Contributing Style Guidelines on Boolean Arguments](https://github.com/apache/iceberg/blob/master/CONTRIBUTING.md#boolean-arguments)
##########
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")
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1203054834
##########
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")
+ public void testAggregatePushDownForIncrementalScan() {
+ sql("CREATE TABLE %s (id LONG, data INT) USING iceberg", tableName);
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, 1111), (1, 2222), (2, 3333), (2, 4444), (3, 5555), (3, 6666) ",
+ tableName);
+ long snapshotId1 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (4, 7777), (5, 8888)", tableName);
+ long snapshotId2 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (6, -7777), (7, 8888)", tableName);
+ long snapshotId3 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (8, 7777), (9, 9999)", tableName);
+
+ Dataset<Row> dfWithAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain1 =
+ dfWithAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ boolean explainContainsPushDownAggregates1 = false;
+ if (explain1.contains("count(data)")
+ && explain1.contains("min(data)")
+ && explain1.contains("max(data)")) {
+ explainContainsPushDownAggregates1 = true;
+ }
+
+ Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates1);
+
+ Dataset<Row> dfWithoutAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .option(SparkReadOptions.AGGREGATE_PUSH_DOWN_ENABLED, "false")
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain2 =
+ dfWithoutAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ explainContainsPushDownAggregates1 = false;
Review Comment:
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` .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1204406703
##########
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")
+ public void testAggregatePushDownForIncrementalScan() {
+ sql("CREATE TABLE %s (id LONG, data INT) USING iceberg", tableName);
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, 1111), (1, 2222), (2, 3333), (2, 4444), (3, 5555), (3, 6666) ",
+ tableName);
+ long snapshotId1 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (4, 7777), (5, 8888)", tableName);
+ long snapshotId2 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (6, -7777), (7, 8888)", tableName);
+ long snapshotId3 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (8, 7777), (9, 9999)", tableName);
+
+ Dataset<Row> dfWithAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain1 =
+ dfWithAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ boolean explainContainsPushDownAggregates1 = false;
+ if (explain1.contains("count(data)")
+ && explain1.contains("min(data)")
+ && explain1.contains("max(data)")) {
+ explainContainsPushDownAggregates1 = true;
+ }
+
+ Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates1);
+
+ Dataset<Row> dfWithoutAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .option(SparkReadOptions.AGGREGATE_PUSH_DOWN_ENABLED, "false")
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain2 =
+ dfWithoutAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ explainContainsPushDownAggregates1 = false;
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1204426716
##########
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 =
Review Comment:
batchResult
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1204425862
##########
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")
Review Comment:
Added some suggested renames, but would also be good to just split the use cases into separate tests to avoid having complicated variable names
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1199293716
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -376,6 +363,12 @@ public Scan build() {
}
private Scan buildBatchScan() {
+ Schema expectedSchema = schemaWithMetadataColumns();
+ return new SparkBatchQueryScan(
+ spark, table, buildIcebergBatchScan(false), readConf, expectedSchema, filterExpressions);
+ }
+
+ private org.apache.iceberg.Scan buildIcebergBatchScan(boolean withStats) {
Review Comment:
nit: Do we need the fully qualified type?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on pull request #7636: Support Aggregate push down for incremental scan
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#issuecomment-1635061515
@RussellSpitzer was this close, and do you think it should go to 1.3.1?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#issuecomment-1669908297
The conflict is very small so i think we can just merge after we get that fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on pull request #7636: Support Aggregate push down for incremental scan
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#issuecomment-1552255265
cc @aokolnychyi @RussellSpitzer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1199560264
##########
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 =
spark
.read()
.format("iceberg")
.option("start-snapshot-id", snapshotIds.get(3).toString())
- .load(tableLocation)
- .orderBy("id")
- .as(Encoders.bean(SimpleRecord.class))
- .collectAsList();
- Assert.assertEquals("Records should match", expectedRecords.subList(1, 4), result);
+ .load(tableLocation);
+ List<SimpleRecord> result1 =
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] ChristinaTech commented on pull request #7636: Support Aggregate push down for incremental scan
Posted by "ChristinaTech (via GitHub)" <gi...@apache.org>.
ChristinaTech commented on PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#issuecomment-1669861230
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on pull request #7636: Support Aggregate push down for incremental scan
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#issuecomment-1552255731
also cc @amogh-jahagirdar @singhpk234 @nastra @ChristinaTech
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1198001598
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -221,15 +222,26 @@ public boolean pushAggregation(Aggregation aggregation) {
return false;
}
- TableScan scan = table.newScan().includeColumnStats();
- Snapshot snapshot = readSnapshot();
- if (snapshot == null) {
- LOG.info("Skipping aggregate pushdown: table snapshot is null");
- return false;
+ org.apache.iceberg.Scan scan;
Review Comment:
minor style thought?
Could we do something like
```java
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
```
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1202922603
##########
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")
+ public void testAggregatePushDownForIncrementalScan() {
+ sql("CREATE TABLE %s (id LONG, data INT) USING iceberg", tableName);
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, 1111), (1, 2222), (2, 3333), (2, 4444), (3, 5555), (3, 6666) ",
+ tableName);
+ long snapshotId1 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (4, 7777), (5, 8888)", tableName);
+ long snapshotId2 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (6, -7777), (7, 8888)", tableName);
+ long snapshotId3 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (8, 7777), (9, 9999)", tableName);
+
+ Dataset<Row> dfWithAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain1 =
+ dfWithAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ boolean explainContainsPushDownAggregates1 = false;
+ if (explain1.contains("count(data)")
+ && explain1.contains("min(data)")
+ && explain1.contains("max(data)")) {
+ explainContainsPushDownAggregates1 = true;
+ }
+
+ Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates1);
+
+ Dataset<Row> dfWithoutAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .option(SparkReadOptions.AGGREGATE_PUSH_DOWN_ENABLED, "false")
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain2 =
+ dfWithoutAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ explainContainsPushDownAggregates1 = false;
Review Comment:
Also how do we make sure that the function expressions are in the TableRelation and not somewhere else in the plan?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1202919345
##########
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")
+ public void testAggregatePushDownForIncrementalScan() {
+ sql("CREATE TABLE %s (id LONG, data INT) USING iceberg", tableName);
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, 1111), (1, 2222), (2, 3333), (2, 4444), (3, 5555), (3, 6666) ",
+ tableName);
+ long snapshotId1 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (4, 7777), (5, 8888)", tableName);
+ long snapshotId2 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (6, -7777), (7, 8888)", tableName);
+ long snapshotId3 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (8, 7777), (9, 9999)", tableName);
+
+ Dataset<Row> dfWithAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain1 =
+ dfWithAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ boolean explainContainsPushDownAggregates1 = false;
+ if (explain1.contains("count(data)")
+ && explain1.contains("min(data)")
+ && explain1.contains("max(data)")) {
+ explainContainsPushDownAggregates1 = true;
+ }
+
+ Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates1);
+
+ Dataset<Row> dfWithoutAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .option(SparkReadOptions.AGGREGATE_PUSH_DOWN_ENABLED, "false")
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain2 =
+ dfWithoutAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ explainContainsPushDownAggregates1 = false;
Review Comment:
We have a StringAssert.contains(strings*) you can use (which will also give a nicer exception)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1204424734
##########
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")
+ public void testAggregatePushDownForIncrementalScan() {
+ sql("CREATE TABLE %s (id LONG, data INT) USING iceberg", tableName);
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, 1111), (1, 2222), (2, 3333), (2, 4444), (3, 5555), (3, 6666) ",
+ tableName);
+ long snapshotId1 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (4, 7777), (5, 8888)", tableName);
+ long snapshotId2 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (6, -7777), (7, 8888)", tableName);
+ long snapshotId3 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (8, 7777), (9, 9999)", tableName);
+
+ Dataset<Row> dfWithAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain1 =
+ dfWithAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ boolean explainContainsPushDownAggregates1 = false;
+ if (explain1.contains("count(data)")
+ && explain1.contains("min(data)")
+ && explain1.contains("max(data)")) {
+ explainContainsPushDownAggregates1 = true;
+ }
+
+ Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates1);
+
+ Dataset<Row> dfWithoutAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .option(SparkReadOptions.AGGREGATE_PUSH_DOWN_ENABLED, "false")
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain2 =
+ dfWithoutAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ explainContainsPushDownAggregates1 = false;
+ if (explain2.contains("count(data)")
+ || explain2.contains("min(data)")
+ || explain2.contains("max(data)")) {
+ explainContainsPushDownAggregates1 = true;
+ }
+
+ Assert.assertFalse("aggregate pushed down", explainContainsPushDownAggregates1);
+ assertEquals(
+ "Aggregate pushdown and non-aggregate pushdown should have the same results",
+ rowsToJava(dfWithAggPushdown1.collectAsList()),
+ rowsToJava(dfWithoutAggPushdown1.collectAsList()));
+
+ Dataset<Row> dfWithAggPushdown2 =
Review Comment:
unboundedPushdownResult
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #7636: Support Aggregate push down for incremental scan
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #7636:
URL: https://github.com/apache/iceberg/pull/7636#discussion_r1204424131
##########
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")
+ public void testAggregatePushDownForIncrementalScan() {
+ sql("CREATE TABLE %s (id LONG, data INT) USING iceberg", tableName);
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, 1111), (1, 2222), (2, 3333), (2, 4444), (3, 5555), (3, 6666) ",
+ tableName);
+ long snapshotId1 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (4, 7777), (5, 8888)", tableName);
+ long snapshotId2 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (6, -7777), (7, 8888)", tableName);
+ long snapshotId3 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (8, 7777), (9, 9999)", tableName);
+
+ Dataset<Row> dfWithAggPushdown1 =
Review Comment:
pushdownResult
##########
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")
+ public void testAggregatePushDownForIncrementalScan() {
+ sql("CREATE TABLE %s (id LONG, data INT) USING iceberg", tableName);
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, 1111), (1, 2222), (2, 3333), (2, 4444), (3, 5555), (3, 6666) ",
+ tableName);
+ long snapshotId1 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (4, 7777), (5, 8888)", tableName);
+ long snapshotId2 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (6, -7777), (7, 8888)", tableName);
+ long snapshotId3 = validationCatalog.loadTable(tableIdent).currentSnapshot().snapshotId();
+ sql("INSERT INTO %s VALUES (8, 7777), (9, 9999)", tableName);
+
+ Dataset<Row> dfWithAggPushdown1 =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.START_SNAPSHOT_ID, snapshotId2)
+ .option(SparkReadOptions.END_SNAPSHOT_ID, snapshotId3)
+ .load(tableName)
+ .agg(functions.min("data"), functions.max("data"), functions.count("data"));
+ String explain1 =
+ dfWithAggPushdown1.queryExecution().explainString(ExplainMode.fromString("simple"));
+ boolean explainContainsPushDownAggregates1 = false;
+ if (explain1.contains("count(data)")
+ && explain1.contains("min(data)")
+ && explain1.contains("max(data)")) {
+ explainContainsPushDownAggregates1 = true;
+ }
+
+ Assert.assertTrue("aggregate pushed down", explainContainsPushDownAggregates1);
+
+ Dataset<Row> dfWithoutAggPushdown1 =
Review Comment:
noPushdownResult
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org