You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "naveenp2708 (via GitHub)" <gi...@apache.org> on 2026/10/05 15:33:17 UTC
[PR] Spark 4.2: Add an SPJ test for an aggregate over a partially clustered join [iceberg]
naveenp2708 opened a new pull request, #18383:
URL: https://github.com/apache/iceberg/pull/18383
Adds a test for a storage-partitioned join where a `DISTINCT` or `GROUP BY` runs directly on a partially clustered join output. The existing `testAggregates` only covers an aggregate over a single scan, not over a join, so this fills that gap.
Before SPARK-55848 this case silently returned wrong counts (1200 rows instead of 200, multiplied by the partition split factor). It fails on Spark 4.0.2 and passes on 4.0.3, which is the fix version. Test only, no production change.
One question: the SPJ tests here run with partial clustering off right now, so I wasn't sure if this coverage belongs here or if you consider it Spark side. Happy to move or drop it if you'd rather.
--
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
Re: [PR] Spark 4.2: Add an SPJ test for an aggregate over a partially clustered join [iceberg]
Posted by "huan233usc (via GitHub)" <gi...@apache.org>.
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: 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