This is an automated email from the ASF dual-hosted git repository.
huaxingao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 87a7e4b134 Spark: Also disable min/max aggregation push down for
binary (#16328)
87a7e4b134 is described below
commit 87a7e4b13470b238e86872e5b194b80c3a89a80f
Author: Dong Wang <[email protected]>
AuthorDate: Thu May 14 23:35:00 2026 +0800
Spark: Also disable min/max aggregation push down for binary (#16328)
Co-authored-by: Timothy Meehan <[email protected]>
---
.../iceberg/spark/source/SparkScanBuilder.java | 3 +-
.../iceberg/spark/sql/TestAggregatePushDown.java | 86 +++++++++++++++++++---
.../iceberg/spark/source/SparkScanBuilder.java | 3 +-
.../iceberg/spark/sql/TestAggregatePushDown.java | 86 +++++++++++++++++++---
.../iceberg/spark/source/SparkScanBuilder.java | 3 +-
.../iceberg/spark/sql/TestAggregatePushDown.java | 86 +++++++++++++++++++---
.../iceberg/spark/source/SparkScanBuilder.java | 3 +-
.../iceberg/spark/sql/TestAggregatePushDown.java | 86 +++++++++++++++++++---
8 files changed, 312 insertions(+), 44 deletions(-)
diff --git
a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
index 3e26da4662..8b75906a62 100644
---
a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
+++
b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
@@ -307,7 +307,8 @@ public class SparkScanBuilder
colName);
return false;
}
- } else if (aggregate.type().typeId() == Type.TypeID.STRING) {
+ } else if (aggregate.type().typeId() == Type.TypeID.STRING
+ || aggregate.type().typeId() == Type.TypeID.BINARY) {
// lower_bounds and upper_bounds may have been truncated before, so
disable push down
// regardless of the current mode
if (aggregate.op() == Expression.Operation.MAX
diff --git
a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
index 462cc5064e..646e96eb54 100644
---
a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
+++
b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
@@ -124,8 +124,7 @@ public class TestAggregatePushDown extends CatalogTestBase {
+ "max(boolean_data), min(boolean_data), count(boolean_data), "
+ "max(float_data), min(float_data), count(float_data), "
+ "max(double_data), min(double_data), count(double_data), "
- + "max(decimal_data), min(decimal_data), count(decimal_data), "
- + "max(binary_data), min(binary_data), count(binary_data) FROM %s";
+ + "max(decimal_data), min(decimal_data), count(decimal_data) FROM
%s";
List<Object[]> explain = sql("EXPLAIN " + select, tableName);
String explainString =
explain.get(0)[0].toString().toLowerCase(Locale.ROOT);
@@ -148,10 +147,7 @@ public class TestAggregatePushDown extends CatalogTestBase
{
&& explainString.contains("count(double_data)")
&& explainString.contains("max(decimal_data)")
&& explainString.contains("min(decimal_data)")
- && explainString.contains("count(decimal_data)")
- && explainString.contains("max(binary_data)")
- && explainString.contains("min(binary_data)")
- && explainString.contains("count(binary_data)")) {
+ && explainString.contains("count(decimal_data)")) {
explainContainsPushDownAggregates = true;
}
@@ -181,10 +177,7 @@ public class TestAggregatePushDown extends CatalogTestBase
{
5L,
new BigDecimal("66.66"),
new BigDecimal("11.11"),
- 6L,
- new byte[] {85, 85},
- new byte[] {17, 17},
- 5L
+ 6L
});
assertEquals("min/max/count push down", expected, actual);
}
@@ -396,6 +389,79 @@ public class TestAggregatePushDown extends CatalogTestBase
{
assertEquals("expected and actual should equal", expected3, actual3);
}
+ @TestTemplate
+ public void testAggregateNotPushDownForBinaryType() {
+ sql("CREATE TABLE %s (id LONG, data BINARY) USING iceberg", tableName);
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "truncate(5)");
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, X'111111111111'), (1, X'2222'),"
+ + " (2, X'3333'), (2, X'4444'),"
+ + " (3, X'5555'), (3, X'555555555555') ",
+ tableName);
+
+ String select1 = "SELECT MAX(id), MAX(data), MIN(data) FROM %s";
+
+ List<Object[]> explain1 = sql("EXPLAIN " + select1, tableName);
+ String explainString1 =
explain1.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ boolean explainContainsPushDownAggregates = false;
+ if (explainString1.contains("max(id)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should not contain the pushed down aggregates")
+ .isFalse();
+
+ List<Object[]> actual1 = sql(select1, tableName);
+ List<Object[]> expected1 = Lists.newArrayList();
+ expected1.add(
+ new Object[] {
+ 3L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17,
17, 17}
+ });
+ assertEquals("expected and actual should equal", expected1, actual1);
+
+ String select2 = "SELECT COUNT(data) FROM %s";
+ List<Object[]> explain2 = sql("EXPLAIN " + select2, tableName);
+ String explainString2 =
explain2.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ if (explainString2.contains("count(data)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should contain the pushed down aggregates")
+ .isTrue();
+
+ List<Object[]> actual2 = sql(select2, tableName);
+ List<Object[]> expected2 = Lists.newArrayList();
+ expected2.add(new Object[] {6L});
+ assertEquals("expected and actual should equal", expected2, actual2);
+
+ explainContainsPushDownAggregates = false;
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "full");
+ String select3 = "SELECT count(data), max(data), min(data) FROM %s";
+ List<Object[]> explain3 = sql("EXPLAIN " + select3, tableName);
+ String explainString3 =
explain3.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ if (explainString3.contains("count(data)") &&
explainString3.contains("max(data)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should not contain the pushed down aggregates")
+ .isFalse();
+
+ List<Object[]> actual3 = sql(select3, tableName);
+ List<Object[]> expected3 = Lists.newArrayList();
+ expected3.add(
+ new Object[] {
+ 6L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17,
17, 17}
+ });
+ assertEquals("expected and actual should equal", expected3, actual3);
+ }
+
@TestTemplate
public void testAggregatePushDownWithDataFilter() {
testAggregatePushDownWithFilter(false);
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
index 3e26da4662..8b75906a62 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
@@ -307,7 +307,8 @@ public class SparkScanBuilder
colName);
return false;
}
- } else if (aggregate.type().typeId() == Type.TypeID.STRING) {
+ } else if (aggregate.type().typeId() == Type.TypeID.STRING
+ || aggregate.type().typeId() == Type.TypeID.BINARY) {
// lower_bounds and upper_bounds may have been truncated before, so
disable push down
// regardless of the current mode
if (aggregate.op() == Expression.Operation.MAX
diff --git
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
index 462cc5064e..646e96eb54 100644
---
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
+++
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
@@ -124,8 +124,7 @@ public class TestAggregatePushDown extends CatalogTestBase {
+ "max(boolean_data), min(boolean_data), count(boolean_data), "
+ "max(float_data), min(float_data), count(float_data), "
+ "max(double_data), min(double_data), count(double_data), "
- + "max(decimal_data), min(decimal_data), count(decimal_data), "
- + "max(binary_data), min(binary_data), count(binary_data) FROM %s";
+ + "max(decimal_data), min(decimal_data), count(decimal_data) FROM
%s";
List<Object[]> explain = sql("EXPLAIN " + select, tableName);
String explainString =
explain.get(0)[0].toString().toLowerCase(Locale.ROOT);
@@ -148,10 +147,7 @@ public class TestAggregatePushDown extends CatalogTestBase
{
&& explainString.contains("count(double_data)")
&& explainString.contains("max(decimal_data)")
&& explainString.contains("min(decimal_data)")
- && explainString.contains("count(decimal_data)")
- && explainString.contains("max(binary_data)")
- && explainString.contains("min(binary_data)")
- && explainString.contains("count(binary_data)")) {
+ && explainString.contains("count(decimal_data)")) {
explainContainsPushDownAggregates = true;
}
@@ -181,10 +177,7 @@ public class TestAggregatePushDown extends CatalogTestBase
{
5L,
new BigDecimal("66.66"),
new BigDecimal("11.11"),
- 6L,
- new byte[] {85, 85},
- new byte[] {17, 17},
- 5L
+ 6L
});
assertEquals("min/max/count push down", expected, actual);
}
@@ -396,6 +389,79 @@ public class TestAggregatePushDown extends CatalogTestBase
{
assertEquals("expected and actual should equal", expected3, actual3);
}
+ @TestTemplate
+ public void testAggregateNotPushDownForBinaryType() {
+ sql("CREATE TABLE %s (id LONG, data BINARY) USING iceberg", tableName);
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "truncate(5)");
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, X'111111111111'), (1, X'2222'),"
+ + " (2, X'3333'), (2, X'4444'),"
+ + " (3, X'5555'), (3, X'555555555555') ",
+ tableName);
+
+ String select1 = "SELECT MAX(id), MAX(data), MIN(data) FROM %s";
+
+ List<Object[]> explain1 = sql("EXPLAIN " + select1, tableName);
+ String explainString1 =
explain1.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ boolean explainContainsPushDownAggregates = false;
+ if (explainString1.contains("max(id)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should not contain the pushed down aggregates")
+ .isFalse();
+
+ List<Object[]> actual1 = sql(select1, tableName);
+ List<Object[]> expected1 = Lists.newArrayList();
+ expected1.add(
+ new Object[] {
+ 3L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17,
17, 17}
+ });
+ assertEquals("expected and actual should equal", expected1, actual1);
+
+ String select2 = "SELECT COUNT(data) FROM %s";
+ List<Object[]> explain2 = sql("EXPLAIN " + select2, tableName);
+ String explainString2 =
explain2.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ if (explainString2.contains("count(data)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should contain the pushed down aggregates")
+ .isTrue();
+
+ List<Object[]> actual2 = sql(select2, tableName);
+ List<Object[]> expected2 = Lists.newArrayList();
+ expected2.add(new Object[] {6L});
+ assertEquals("expected and actual should equal", expected2, actual2);
+
+ explainContainsPushDownAggregates = false;
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "full");
+ String select3 = "SELECT count(data), max(data), min(data) FROM %s";
+ List<Object[]> explain3 = sql("EXPLAIN " + select3, tableName);
+ String explainString3 =
explain3.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ if (explainString3.contains("count(data)") &&
explainString3.contains("max(data)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should not contain the pushed down aggregates")
+ .isFalse();
+
+ List<Object[]> actual3 = sql(select3, tableName);
+ List<Object[]> expected3 = Lists.newArrayList();
+ expected3.add(
+ new Object[] {
+ 6L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17,
17, 17}
+ });
+ assertEquals("expected and actual should equal", expected3, actual3);
+ }
+
@TestTemplate
public void testAggregatePushDownWithDataFilter() {
testAggregatePushDownWithFilter(false);
diff --git
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
index 3e26da4662..8b75906a62 100644
---
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
+++
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
@@ -307,7 +307,8 @@ public class SparkScanBuilder
colName);
return false;
}
- } else if (aggregate.type().typeId() == Type.TypeID.STRING) {
+ } else if (aggregate.type().typeId() == Type.TypeID.STRING
+ || aggregate.type().typeId() == Type.TypeID.BINARY) {
// lower_bounds and upper_bounds may have been truncated before, so
disable push down
// regardless of the current mode
if (aggregate.op() == Expression.Operation.MAX
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
index e707067980..1669301d2d 100644
---
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
+++
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
@@ -124,8 +124,7 @@ public class TestAggregatePushDown extends CatalogTestBase {
+ "max(boolean_data), min(boolean_data), count(boolean_data), "
+ "max(float_data), min(float_data), count(float_data), "
+ "max(double_data), min(double_data), count(double_data), "
- + "max(decimal_data), min(decimal_data), count(decimal_data), "
- + "max(binary_data), min(binary_data), count(binary_data) FROM %s";
+ + "max(decimal_data), min(decimal_data), count(decimal_data) FROM
%s";
List<Object[]> explain = sql("EXPLAIN " + select, tableName);
String explainString =
explain.get(0)[0].toString().toLowerCase(Locale.ROOT);
@@ -148,10 +147,7 @@ public class TestAggregatePushDown extends CatalogTestBase
{
&& explainString.contains("count(double_data)")
&& explainString.contains("max(decimal_data)")
&& explainString.contains("min(decimal_data)")
- && explainString.contains("count(decimal_data)")
- && explainString.contains("max(binary_data)")
- && explainString.contains("min(binary_data)")
- && explainString.contains("count(binary_data)")) {
+ && explainString.contains("count(decimal_data)")) {
explainContainsPushDownAggregates = true;
}
@@ -181,10 +177,7 @@ public class TestAggregatePushDown extends CatalogTestBase
{
5L,
new BigDecimal("66.66"),
new BigDecimal("11.11"),
- 6L,
- new byte[] {85, 85},
- new byte[] {17, 17},
- 5L
+ 6L
});
assertEquals("min/max/count push down", expected, actual);
}
@@ -396,6 +389,79 @@ public class TestAggregatePushDown extends CatalogTestBase
{
assertEquals("expected and actual should equal", expected3, actual3);
}
+ @TestTemplate
+ public void testAggregateNotPushDownForBinaryType() {
+ sql("CREATE TABLE %s (id LONG, data BINARY) USING iceberg", tableName);
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "truncate(5)");
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, X'111111111111'), (1, X'2222'),"
+ + " (2, X'3333'), (2, X'4444'),"
+ + " (3, X'5555'), (3, X'555555555555') ",
+ tableName);
+
+ String select1 = "SELECT MAX(id), MAX(data), MIN(data) FROM %s";
+
+ List<Object[]> explain1 = sql("EXPLAIN " + select1, tableName);
+ String explainString1 =
explain1.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ boolean explainContainsPushDownAggregates = false;
+ if (explainString1.contains("max(id)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should not contain the pushed down aggregates")
+ .isFalse();
+
+ List<Object[]> actual1 = sql(select1, tableName);
+ List<Object[]> expected1 = Lists.newArrayList();
+ expected1.add(
+ new Object[] {
+ 3L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17,
17, 17}
+ });
+ assertEquals("expected and actual should equal", expected1, actual1);
+
+ String select2 = "SELECT COUNT(data) FROM %s";
+ List<Object[]> explain2 = sql("EXPLAIN " + select2, tableName);
+ String explainString2 =
explain2.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ if (explainString2.contains("count(data)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should contain the pushed down aggregates")
+ .isTrue();
+
+ List<Object[]> actual2 = sql(select2, tableName);
+ List<Object[]> expected2 = Lists.newArrayList();
+ expected2.add(new Object[] {6L});
+ assertEquals("expected and actual should equal", expected2, actual2);
+
+ explainContainsPushDownAggregates = false;
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "full");
+ String select3 = "SELECT count(data), max(data), min(data) FROM %s";
+ List<Object[]> explain3 = sql("EXPLAIN " + select3, tableName);
+ String explainString3 =
explain3.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ if (explainString3.contains("count(data)") &&
explainString3.contains("max(data)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should not contain the pushed down aggregates")
+ .isFalse();
+
+ List<Object[]> actual3 = sql(select3, tableName);
+ List<Object[]> expected3 = Lists.newArrayList();
+ expected3.add(
+ new Object[] {
+ 6L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17,
17, 17}
+ });
+ assertEquals("expected and actual should equal", expected3, actual3);
+ }
+
@TestTemplate
public void testAggregatePushDownWithDataFilter() {
testAggregatePushDownWithFilter(false);
diff --git
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
index ebbae01a70..69b6314a7f 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
+++
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
@@ -223,7 +223,8 @@ public class SparkScanBuilder extends BaseSparkScanBuilder
colName);
return false;
}
- } else if (aggregate.type().typeId() == Type.TypeID.STRING) {
+ } else if (aggregate.type().typeId() == Type.TypeID.STRING
+ || aggregate.type().typeId() == Type.TypeID.BINARY) {
// lower_bounds and upper_bounds may have been truncated before, so
disable push down
// regardless of the current mode
if (aggregate.op() == Expression.Operation.MAX
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
index e707067980..1669301d2d 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
+++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java
@@ -124,8 +124,7 @@ public class TestAggregatePushDown extends CatalogTestBase {
+ "max(boolean_data), min(boolean_data), count(boolean_data), "
+ "max(float_data), min(float_data), count(float_data), "
+ "max(double_data), min(double_data), count(double_data), "
- + "max(decimal_data), min(decimal_data), count(decimal_data), "
- + "max(binary_data), min(binary_data), count(binary_data) FROM %s";
+ + "max(decimal_data), min(decimal_data), count(decimal_data) FROM
%s";
List<Object[]> explain = sql("EXPLAIN " + select, tableName);
String explainString =
explain.get(0)[0].toString().toLowerCase(Locale.ROOT);
@@ -148,10 +147,7 @@ public class TestAggregatePushDown extends CatalogTestBase
{
&& explainString.contains("count(double_data)")
&& explainString.contains("max(decimal_data)")
&& explainString.contains("min(decimal_data)")
- && explainString.contains("count(decimal_data)")
- && explainString.contains("max(binary_data)")
- && explainString.contains("min(binary_data)")
- && explainString.contains("count(binary_data)")) {
+ && explainString.contains("count(decimal_data)")) {
explainContainsPushDownAggregates = true;
}
@@ -181,10 +177,7 @@ public class TestAggregatePushDown extends CatalogTestBase
{
5L,
new BigDecimal("66.66"),
new BigDecimal("11.11"),
- 6L,
- new byte[] {85, 85},
- new byte[] {17, 17},
- 5L
+ 6L
});
assertEquals("min/max/count push down", expected, actual);
}
@@ -396,6 +389,79 @@ public class TestAggregatePushDown extends CatalogTestBase
{
assertEquals("expected and actual should equal", expected3, actual3);
}
+ @TestTemplate
+ public void testAggregateNotPushDownForBinaryType() {
+ sql("CREATE TABLE %s (id LONG, data BINARY) USING iceberg", tableName);
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "truncate(5)");
+ sql(
+ "INSERT INTO TABLE %s VALUES (1, X'111111111111'), (1, X'2222'),"
+ + " (2, X'3333'), (2, X'4444'),"
+ + " (3, X'5555'), (3, X'555555555555') ",
+ tableName);
+
+ String select1 = "SELECT MAX(id), MAX(data), MIN(data) FROM %s";
+
+ List<Object[]> explain1 = sql("EXPLAIN " + select1, tableName);
+ String explainString1 =
explain1.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ boolean explainContainsPushDownAggregates = false;
+ if (explainString1.contains("max(id)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should not contain the pushed down aggregates")
+ .isFalse();
+
+ List<Object[]> actual1 = sql(select1, tableName);
+ List<Object[]> expected1 = Lists.newArrayList();
+ expected1.add(
+ new Object[] {
+ 3L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17,
17, 17}
+ });
+ assertEquals("expected and actual should equal", expected1, actual1);
+
+ String select2 = "SELECT COUNT(data) FROM %s";
+ List<Object[]> explain2 = sql("EXPLAIN " + select2, tableName);
+ String explainString2 =
explain2.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ if (explainString2.contains("count(data)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should contain the pushed down aggregates")
+ .isTrue();
+
+ List<Object[]> actual2 = sql(select2, tableName);
+ List<Object[]> expected2 = Lists.newArrayList();
+ expected2.add(new Object[] {6L});
+ assertEquals("expected and actual should equal", expected2, actual2);
+
+ explainContainsPushDownAggregates = false;
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, TableProperties.DEFAULT_WRITE_METRICS_MODE, "full");
+ String select3 = "SELECT count(data), max(data), min(data) FROM %s";
+ List<Object[]> explain3 = sql("EXPLAIN " + select3, tableName);
+ String explainString3 =
explain3.get(0)[0].toString().toLowerCase(Locale.ROOT);
+ if (explainString3.contains("count(data)") &&
explainString3.contains("max(data)")) {
+ explainContainsPushDownAggregates = true;
+ }
+
+ assertThat(explainContainsPushDownAggregates)
+ .as("explain should not contain the pushed down aggregates")
+ .isFalse();
+
+ List<Object[]> actual3 = sql(select3, tableName);
+ List<Object[]> expected3 = Lists.newArrayList();
+ expected3.add(
+ new Object[] {
+ 6L, new byte[] {85, 85, 85, 85, 85, 85}, new byte[] {17, 17, 17, 17,
17, 17}
+ });
+ assertEquals("expected and actual should equal", expected3, actual3);
+ }
+
@TestTemplate
public void testAggregatePushDownWithDataFilter() {
testAggregatePushDownWithFilter(false);
