Skip to content

Commit 87a7e4b

Browse files
hantangwangdtdcmeehan
hantangwangd
and
tdcmeehan
authored
Spark: Also disable min/max aggregation push down for binary (#16328)
Co-authored-by: Timothy Meehan <tim@timdmeehan.com>
1 parent 294f80d commit 87a7e4b

8 files changed

Lines changed: 312 additions & 44 deletions

File tree

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -307,7 +307,8 @@ private boolean metricsModeSupportsAggregatePushDown(List<BoundAggregate<?, ?>>
307307
colName);
308308
return false;
309309
}
310-
} else if (aggregate.type().typeId() == Type.TypeID.STRING) {
310+
} else if (aggregate.type().typeId() == Type.TypeID.STRING
311+
|| aggregate.type().typeId() == Type.TypeID.BINARY) {
311312
// lower_bounds and upper_bounds may have been truncated before, so disable push down
312313
// regardless of the current mode
313314
if (aggregate.op() == Expression.Operation.MAX

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

Lines changed: 76 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -124,8 +124,7 @@ private void testDifferentDataTypesAggregatePushDown(boolean hasPartitionCol) {
124124
+ "max(boolean_data), min(boolean_data), count(boolean_data), "
125125
+ "max(float_data), min(float_data), count(float_data), "
126126
+ "max(double_data), min(double_data), count(double_data), "
127-
+ "max(decimal_data), min(decimal_data), count(decimal_data), "
128-
+ "max(binary_data), min(binary_data), count(binary_data) FROM %s";
127+
+ "max(decimal_data), min(decimal_data), count(decimal_data) FROM %s";
129128

130129
List<Object[]> explain = sql("EXPLAIN " + select, tableName);
131130
String explainString = explain.get(0)[0].toString().toLowerCase(Locale.ROOT);
@@ -148,10 +147,7 @@ private void testDifferentDataTypesAggregatePushDown(boolean hasPartitionCol) {
148147
&& explainString.contains("count(double_data)")
149148
&& explainString.contains("max(decimal_data)")
150149
&& explainString.contains("min(decimal_data)")
151-
&& explainString.contains("count(decimal_data)")
152-
&& explainString.contains("max(binary_data)")
153-
&& explainString.contains("min(binary_data)")
154-
&& explainString.contains("count(binary_data)")) {
150+
&& explainString.contains("count(decimal_data)")) {
155151
explainContainsPushDownAggregates = true;
156152
}
157153

@@ -181,10 +177,7 @@ private void testDifferentDataTypesAggregatePushDown(boolean hasPartitionCol) {
181177
5L,
182178
new BigDecimal("66.66"),
183179
new BigDecimal("11.11"),
184-
6L,
185-
new byte[] {85, 85},
186-
new byte[] {17, 17},
187-
5L
180+
6L
188181
});
189182
assertEquals("min/max/count push down", expected, actual);
190183
}
@@ -396,6 +389,79 @@ public void testAggregateNotPushDownForStringType() {
396389
assertEquals("expected and actual should equal", expected3, actual3);
397390
}
398391

392+
@TestTemplate
393+
public void testAggregateNotPushDownForBinaryType() {
394+
sql("CREATE TABLE %s (id LONG, data BINARY) USING iceberg", tableName);
395+
sql(
396+
"ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
397+
tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "truncate(5)");
398+
sql(
399+
"INSERT INTO TABLE %s VALUES (1, X'111111111111'), (1, X'2222'),"
400+
+ " (2, X'3333'), (2, X'4444'),"
401+
+ " (3, X'5555'), (3, X'555555555555') ",
402+
tableName);
403+
404+
String select1 = "SELECT MAX(id), MAX(data), MIN(data) FROM %s";
405+
406+
List<Object[]> explain1 = sql("EXPLAIN " + select1, tableName);
407+
String explainString1 = explain1.get(0)[0].toString().toLowerCase(Locale.ROOT);
408+
boolean explainContainsPushDownAggregates = false;
409+
if (explainString1.contains("max(id)")) {
410+
explainContainsPushDownAggregates = true;
411+
}
412+
413+
assertThat(explainContainsPushDownAggregates)
414+
.as("explain should not contain the pushed down aggregates")
415+
.isFalse();
416+
417+
List<Object[]> actual1 = sql(select1, tableName);
418+
List<Object[]> expected1 = Lists.newArrayList();
419+
expected1.add(
420+
new Object[] {
421+
3L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17, 17, 17}
422+
});
423+
assertEquals("expected and actual should equal", expected1, actual1);
424+
425+
String select2 = "SELECT COUNT(data) FROM %s";
426+
List<Object[]> explain2 = sql("EXPLAIN " + select2, tableName);
427+
String explainString2 = explain2.get(0)[0].toString().toLowerCase(Locale.ROOT);
428+
if (explainString2.contains("count(data)")) {
429+
explainContainsPushDownAggregates = true;
430+
}
431+
432+
assertThat(explainContainsPushDownAggregates)
433+
.as("explain should contain the pushed down aggregates")
434+
.isTrue();
435+
436+
List<Object[]> actual2 = sql(select2, tableName);
437+
List<Object[]> expected2 = Lists.newArrayList();
438+
expected2.add(new Object[] {6L});
439+
assertEquals("expected and actual should equal", expected2, actual2);
440+
441+
explainContainsPushDownAggregates = false;
442+
sql(
443+
"ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
444+
tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "full");
445+
String select3 = "SELECT count(data), max(data), min(data) FROM %s";
446+
List<Object[]> explain3 = sql("EXPLAIN " + select3, tableName);
447+
String explainString3 = explain3.get(0)[0].toString().toLowerCase(Locale.ROOT);
448+
if (explainString3.contains("count(data)") && explainString3.contains("max(data)")) {
449+
explainContainsPushDownAggregates = true;
450+
}
451+
452+
assertThat(explainContainsPushDownAggregates)
453+
.as("explain should not contain the pushed down aggregates")
454+
.isFalse();
455+
456+
List<Object[]> actual3 = sql(select3, tableName);
457+
List<Object[]> expected3 = Lists.newArrayList();
458+
expected3.add(
459+
new Object[] {
460+
6L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17, 17, 17}
461+
});
462+
assertEquals("expected and actual should equal", expected3, actual3);
463+
}
464+
399465
@TestTemplate
400466
public void testAggregatePushDownWithDataFilter() {
401467
testAggregatePushDownWithFilter(false);

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -307,7 +307,8 @@ private boolean metricsModeSupportsAggregatePushDown(List<BoundAggregate<?, ?>>
307307
colName);
308308
return false;
309309
}
310-
} else if (aggregate.type().typeId() == Type.TypeID.STRING) {
310+
} else if (aggregate.type().typeId() == Type.TypeID.STRING
311+
|| aggregate.type().typeId() == Type.TypeID.BINARY) {
311312
// lower_bounds and upper_bounds may have been truncated before, so disable push down
312313
// regardless of the current mode
313314
if (aggregate.op() == Expression.Operation.MAX

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

Lines changed: 76 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -124,8 +124,7 @@ private void testDifferentDataTypesAggregatePushDown(boolean hasPartitionCol) {
124124
+ "max(boolean_data), min(boolean_data), count(boolean_data), "
125125
+ "max(float_data), min(float_data), count(float_data), "
126126
+ "max(double_data), min(double_data), count(double_data), "
127-
+ "max(decimal_data), min(decimal_data), count(decimal_data), "
128-
+ "max(binary_data), min(binary_data), count(binary_data) FROM %s";
127+
+ "max(decimal_data), min(decimal_data), count(decimal_data) FROM %s";
129128

130129
List<Object[]> explain = sql("EXPLAIN " + select, tableName);
131130
String explainString = explain.get(0)[0].toString().toLowerCase(Locale.ROOT);
@@ -148,10 +147,7 @@ private void testDifferentDataTypesAggregatePushDown(boolean hasPartitionCol) {
148147
&& explainString.contains("count(double_data)")
149148
&& explainString.contains("max(decimal_data)")
150149
&& explainString.contains("min(decimal_data)")
151-
&& explainString.contains("count(decimal_data)")
152-
&& explainString.contains("max(binary_data)")
153-
&& explainString.contains("min(binary_data)")
154-
&& explainString.contains("count(binary_data)")) {
150+
&& explainString.contains("count(decimal_data)")) {
155151
explainContainsPushDownAggregates = true;
156152
}
157153

@@ -181,10 +177,7 @@ private void testDifferentDataTypesAggregatePushDown(boolean hasPartitionCol) {
181177
5L,
182178
new BigDecimal("66.66"),
183179
new BigDecimal("11.11"),
184-
6L,
185-
new byte[] {85, 85},
186-
new byte[] {17, 17},
187-
5L
180+
6L
188181
});
189182
assertEquals("min/max/count push down", expected, actual);
190183
}
@@ -396,6 +389,79 @@ public void testAggregateNotPushDownForStringType() {
396389
assertEquals("expected and actual should equal", expected3, actual3);
397390
}
398391

392+
@TestTemplate
393+
public void testAggregateNotPushDownForBinaryType() {
394+
sql("CREATE TABLE %s (id LONG, data BINARY) USING iceberg", tableName);
395+
sql(
396+
"ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
397+
tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "truncate(5)");
398+
sql(
399+
"INSERT INTO TABLE %s VALUES (1, X'111111111111'), (1, X'2222'),"
400+
+ " (2, X'3333'), (2, X'4444'),"
401+
+ " (3, X'5555'), (3, X'555555555555') ",
402+
tableName);
403+
404+
String select1 = "SELECT MAX(id), MAX(data), MIN(data) FROM %s";
405+
406+
List<Object[]> explain1 = sql("EXPLAIN " + select1, tableName);
407+
String explainString1 = explain1.get(0)[0].toString().toLowerCase(Locale.ROOT);
408+
boolean explainContainsPushDownAggregates = false;
409+
if (explainString1.contains("max(id)")) {
410+
explainContainsPushDownAggregates = true;
411+
}
412+
413+
assertThat(explainContainsPushDownAggregates)
414+
.as("explain should not contain the pushed down aggregates")
415+
.isFalse();
416+
417+
List<Object[]> actual1 = sql(select1, tableName);
418+
List<Object[]> expected1 = Lists.newArrayList();
419+
expected1.add(
420+
new Object[] {
421+
3L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17, 17, 17}
422+
});
423+
assertEquals("expected and actual should equal", expected1, actual1);
424+
425+
String select2 = "SELECT COUNT(data) FROM %s";
426+
List<Object[]> explain2 = sql("EXPLAIN " + select2, tableName);
427+
String explainString2 = explain2.get(0)[0].toString().toLowerCase(Locale.ROOT);
428+
if (explainString2.contains("count(data)")) {
429+
explainContainsPushDownAggregates = true;
430+
}
431+
432+
assertThat(explainContainsPushDownAggregates)
433+
.as("explain should contain the pushed down aggregates")
434+
.isTrue();
435+
436+
List<Object[]> actual2 = sql(select2, tableName);
437+
List<Object[]> expected2 = Lists.newArrayList();
438+
expected2.add(new Object[] {6L});
439+
assertEquals("expected and actual should equal", expected2, actual2);
440+
441+
explainContainsPushDownAggregates = false;
442+
sql(
443+
"ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
444+
tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "full");
445+
String select3 = "SELECT count(data), max(data), min(data) FROM %s";
446+
List<Object[]> explain3 = sql("EXPLAIN " + select3, tableName);
447+
String explainString3 = explain3.get(0)[0].toString().toLowerCase(Locale.ROOT);
448+
if (explainString3.contains("count(data)") && explainString3.contains("max(data)")) {
449+
explainContainsPushDownAggregates = true;
450+
}
451+
452+
assertThat(explainContainsPushDownAggregates)
453+
.as("explain should not contain the pushed down aggregates")
454+
.isFalse();
455+
456+
List<Object[]> actual3 = sql(select3, tableName);
457+
List<Object[]> expected3 = Lists.newArrayList();
458+
expected3.add(
459+
new Object[] {
460+
6L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17, 17, 17}
461+
});
462+
assertEquals("expected and actual should equal", expected3, actual3);
463+
}
464+
399465
@TestTemplate
400466
public void testAggregatePushDownWithDataFilter() {
401467
testAggregatePushDownWithFilter(false);

‎spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -307,7 +307,8 @@ private boolean metricsModeSupportsAggregatePushDown(List<BoundAggregate<?, ?>>
307307
colName);
308308
return false;
309309
}
310-
} else if (aggregate.type().typeId() == Type.TypeID.STRING) {
310+
} else if (aggregate.type().typeId() == Type.TypeID.STRING
311+
|| aggregate.type().typeId() == Type.TypeID.BINARY) {
311312
// lower_bounds and upper_bounds may have been truncated before, so disable push down
312313
// regardless of the current mode
313314
if (aggregate.op() == Expression.Operation.MAX

‎spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java‎

Lines changed: 76 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -124,8 +124,7 @@ private void testDifferentDataTypesAggregatePushDown(boolean hasPartitionCol) {
124124
+ "max(boolean_data), min(boolean_data), count(boolean_data), "
125125
+ "max(float_data), min(float_data), count(float_data), "
126126
+ "max(double_data), min(double_data), count(double_data), "
127-
+ "max(decimal_data), min(decimal_data), count(decimal_data), "
128-
+ "max(binary_data), min(binary_data), count(binary_data) FROM %s";
127+
+ "max(decimal_data), min(decimal_data), count(decimal_data) FROM %s";
129128

130129
List<Object[]> explain = sql("EXPLAIN " + select, tableName);
131130
String explainString = explain.get(0)[0].toString().toLowerCase(Locale.ROOT);
@@ -148,10 +147,7 @@ private void testDifferentDataTypesAggregatePushDown(boolean hasPartitionCol) {
148147
&& explainString.contains("count(double_data)")
149148
&& explainString.contains("max(decimal_data)")
150149
&& explainString.contains("min(decimal_data)")
151-
&& explainString.contains("count(decimal_data)")
152-
&& explainString.contains("max(binary_data)")
153-
&& explainString.contains("min(binary_data)")
154-
&& explainString.contains("count(binary_data)")) {
150+
&& explainString.contains("count(decimal_data)")) {
155151
explainContainsPushDownAggregates = true;
156152
}
157153

@@ -181,10 +177,7 @@ private void testDifferentDataTypesAggregatePushDown(boolean hasPartitionCol) {
181177
5L,
182178
new BigDecimal("66.66"),
183179
new BigDecimal("11.11"),
184-
6L,
185-
new byte[] {85, 85},
186-
new byte[] {17, 17},
187-
5L
180+
6L
188181
});
189182
assertEquals("min/max/count push down", expected, actual);
190183
}
@@ -396,6 +389,79 @@ public void testAggregateNotPushDownForStringType() {
396389
assertEquals("expected and actual should equal", expected3, actual3);
397390
}
398391

392+
@TestTemplate
393+
public void testAggregateNotPushDownForBinaryType() {
394+
sql("CREATE TABLE %s (id LONG, data BINARY) USING iceberg", tableName);
395+
sql(
396+
"ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
397+
tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "truncate(5)");
398+
sql(
399+
"INSERT INTO TABLE %s VALUES (1, X'111111111111'), (1, X'2222'),"
400+
+ " (2, X'3333'), (2, X'4444'),"
401+
+ " (3, X'5555'), (3, X'555555555555') ",
402+
tableName);
403+
404+
String select1 = "SELECT MAX(id), MAX(data), MIN(data) FROM %s";
405+
406+
List<Object[]> explain1 = sql("EXPLAIN " + select1, tableName);
407+
String explainString1 = explain1.get(0)[0].toString().toLowerCase(Locale.ROOT);
408+
boolean explainContainsPushDownAggregates = false;
409+
if (explainString1.contains("max(id)")) {
410+
explainContainsPushDownAggregates = true;
411+
}
412+
413+
assertThat(explainContainsPushDownAggregates)
414+
.as("explain should not contain the pushed down aggregates")
415+
.isFalse();
416+
417+
List<Object[]> actual1 = sql(select1, tableName);
418+
List<Object[]> expected1 = Lists.newArrayList();
419+
expected1.add(
420+
new Object[] {
421+
3L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17, 17, 17}
422+
});
423+
assertEquals("expected and actual should equal", expected1, actual1);
424+
425+
String select2 = "SELECT COUNT(data) FROM %s";
426+
List<Object[]> explain2 = sql("EXPLAIN " + select2, tableName);
427+
String explainString2 = explain2.get(0)[0].toString().toLowerCase(Locale.ROOT);
428+
if (explainString2.contains("count(data)")) {
429+
explainContainsPushDownAggregates = true;
430+
}
431+
432+
assertThat(explainContainsPushDownAggregates)
433+
.as("explain should contain the pushed down aggregates")
434+
.isTrue();
435+
436+
List<Object[]> actual2 = sql(select2, tableName);
437+
List<Object[]> expected2 = Lists.newArrayList();
438+
expected2.add(new Object[] {6L});
439+
assertEquals("expected and actual should equal", expected2, actual2);
440+
441+
explainContainsPushDownAggregates = false;
442+
sql(
443+
"ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
444+
tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "full");
445+
String select3 = "SELECT count(data), max(data), min(data) FROM %s";
446+
List<Object[]> explain3 = sql("EXPLAIN " + select3, tableName);
447+
String explainString3 = explain3.get(0)[0].toString().toLowerCase(Locale.ROOT);
448+
if (explainString3.contains("count(data)") && explainString3.contains("max(data)")) {
449+
explainContainsPushDownAggregates = true;
450+
}
451+
452+
assertThat(explainContainsPushDownAggregates)
453+
.as("explain should not contain the pushed down aggregates")
454+
.isFalse();
455+
456+
List<Object[]> actual3 = sql(select3, tableName);
457+
List<Object[]> expected3 = Lists.newArrayList();
458+
expected3.add(
459+
new Object[] {
460+
6L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17, 17, 17}
461+
});
462+
assertEquals("expected and actual should equal", expected3, actual3);
463+
}
464+
399465
@TestTemplate
400466
public void testAggregatePushDownWithDataFilter() {
401467
testAggregatePushDownWithFilter(false);

‎spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -223,7 +223,8 @@ private boolean metricsModeSupportsAggregatePushDown(List<BoundAggregate<?, ?>>
223223
colName);
224224
return false;
225225
}
226-
} else if (aggregate.type().typeId() == Type.TypeID.STRING) {
226+
} else if (aggregate.type().typeId() == Type.TypeID.STRING
227+
|| aggregate.type().typeId() == Type.TypeID.BINARY) {
227228
// lower_bounds and upper_bounds may have been truncated before, so disable push down
228229
// regardless of the current mode
229230
if (aggregate.op() == Expression.Operation.MAX

0 commit comments

Comments
 (0)