huan233usc commented on code in PR #18383:
URL: https://github.com/apache/iceberg/pull/18383#discussion_r4275757814
##########
spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestStoragePartitionedJoins.java:
##########
@@ -724,6 +725,68 @@ public void testAggregates() throws NoSuchTableException {
tableName(OTHER_TABLE_NAME));
}
+ @TestTemplate
+ public void testAggregateOverPartiallyClusteredJoin() throws
NoSuchTableException {
+ // SPARK-55848: an aggregate placed directly on a partially-clustered
storage-partitioned join
+ // output must return the same rows with SPJ enabled as without it.
Partial clustering splits
+ // the skewed side's partitions, and before SPARK-55848 a DISTINCT / GROUP
BY on that output
+ // trusted the split partitioning and over-counted by the split factor
(1200 rows instead of
+ // 200). The existing testAggregates covers an aggregate over a single
scan, not over a
+ // partially-clustered join, so this pins the join case. It fails on Spark
4.0.2 and passes on
+ // 4.0.3, the SPARK-55848 fix version, through the Iceberg Spark runtime.
+ String createTableStmt =
+ "CREATE TABLE %s (id BIGINT, salary INT)"
+ + "USING iceberg "
+ + "PARTITIONED BY (bucket(8, id))"
+ + "TBLPROPERTIES (%s)";
+
+ sql(createTableStmt, tableName, tablePropsAsString(TABLE_PROPERTIES));
+ configurePlanningMode(tableName, planningMode);
+ sql(createTableStmt, tableName(OTHER_TABLE_NAME),
tablePropsAsString(TABLE_PROPERTIES));
+ configurePlanningMode(tableName(OTHER_TABLE_NAME), planningMode);
+
+ // Skew the probe side: append the same rows six times so each bucket of
the first table holds
+ // six files. With split-per-file (16 MB split size and open-file cost)
and partial clustering
+ // enabled, that side's partitions split while the single-file side is
replicated to match.
+ Table table = validationCatalog.loadTable(tableIdent);
+ Dataset<Row> dataDF = randomDataDF(table.schema(), 200);
+ for (int i = 0; i < 6; i++) {
+ append(tableName, dataDF);
+ }
+ append(tableName(OTHER_TABLE_NAME), dataDF);
+
+ Map<String, String> partiallyClusteredSpjConf =
Maps.newHashMap(ENABLED_SPJ_SQL_CONF);
+ partiallyClusteredSpjConf.put(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED().key(),
"true");
+
+ String distinctAfterJoin =
+ "SELECT count(*) AS c FROM ("
+ + "SELECT DISTINCT t1.id FROM %s t1 INNER JOIN %s t2 ON t1.id =
t2.id)";
+ String aggAfterJoin =
+ "SELECT count(*) AS c FROM ("
+ + "SELECT t1.id, count(*) AS cnt FROM %s t1 INNER JOIN %s t2 ON
t1.id = t2.id "
+ + "GROUP BY t1.id)";
+
+ for (String query : new String[] {distinctAfterJoin, aggAfterJoin}) {
+ AtomicReference<List<Object[]>> rowsWithSPJ = new AtomicReference<>();
+ AtomicReference<List<Object[]>> rowsWithoutSPJ = new AtomicReference<>();
+
+ withSQLConf(
+ partiallyClusteredSpjConf,
+ () -> rowsWithSPJ.set(sql(query, tableName,
tableName(OTHER_TABLE_NAME))));
Review Comment:
Could we also assert that the SPJ-enabled plan actually uses a partially
clustered join? A fallback shuffle join, or SPJ without partial clustering,
could still match the baseline and let this regression test pass. A plan check
similar to `assertPartitioningAwarePlan`, scoped to the join so it allows the
aggregate’s required exchange, would verify that the intended path is exercised.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
