gerashegalov commented on code in PR #18709:
URL: https://github.com/apache/iceberg/pull/18709#discussion_r4535457877
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java:
##########
@@ -233,11 +233,14 @@ protected synchronized List<T> tasks() {
@Override
protected synchronized List<ScanTaskGroup<T>> taskGroups() {
if (taskGroups == null) {
+ long splitSize = adjustSplitSize(tasks(), scan.targetSplitSize());
+ setEffectiveSplitTargetSize(splitSize);
Review Comment:
Thanks for checking both cases. Fixed in 513c06382: the target is now
recorded after planning and is 0 whenever no task groups are planned, at both
the batch and staged planning sites. `effectiveSplitTargetSizeWithoutData` now
covers a table without a snapshot, a filter that prunes every file, and a table
whose data was deleted. Without the fix the pruned case reports 16777216.
##########
spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java:
##########
@@ -44,6 +61,180 @@ public class TestSparkReadMetrics extends
TestBaseWithCatalog {
@AfterEach
public void removeTables() {
sql("DROP TABLE IF EXISTS %s", tableName);
+ sql("DROP TABLE IF EXISTS dim");
+ }
+
+ @TestTemplate
+ void effectiveSplitTargetSize() throws NoSuchTableException, IOException {
+ long tableSplitSize = 128L * 1024 * 1024;
+ long optionSplitSize = 32L * 1024 * 1024;
+ int parallelism = 1000;
+ sql(
+ "CREATE TABLE %s (id BIGINT) USING iceberg TBLPROPERTIES
('read.split.target-size'='%s')",
+ tableName, tableSplitSize);
+ spark.range(100).coalesce(1).writeTo(tableName).append();
+
+ long scanSize = 0;
+ try (CloseableIterable<FileScanTask> tasks =
+ validationCatalog.loadTable(tableIdent).newScan().planFiles()) {
+ for (FileScanTask task : tasks) {
+ scanSize += task.sizeBytes();
+ }
+ }
+ long adaptiveSplitSize = TableScanUtil.adjustSplitSize(scanSize,
parallelism, tableSplitSize);
+ assertThat(adaptiveSplitSize).isNotEqualTo(tableSplitSize);
+
+ // adaptive sizing lowers the table target for a small scan
+ withSQLConf(
+ ImmutableMap.of(
+ SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_ENABLED,
+ "true",
+ SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_PARALLELISM,
+ String.valueOf(parallelism)),
+ () -> {
+ Dataset<Row> df = spark.sql(String.format("SELECT * FROM %s",
tableName));
+ df.collect();
+ SparkPlan scan =
df.queryExecution().executedPlan().collectLeaves().head();
+
assertThat(scan.metrics().get("effectiveSplitTargetSize").get().value())
+ .isEqualTo(adaptiveSplitSize);
+ });
+
+ // without adaptive sizing, the table property is the target
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_ENABLED,
"false"),
+ () -> {
+ Dataset<Row> df = spark.sql(String.format("SELECT * FROM %s",
tableName));
+ df.collect();
+ SparkPlan scan =
df.queryExecution().executedPlan().collectLeaves().head();
+
assertThat(scan.metrics().get("effectiveSplitTargetSize").get().value())
+ .isEqualTo(tableSplitSize);
+ });
+
+ // an explicit read option is used as is, even with adaptive sizing enabled
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_ENABLED,
"true"),
+ () -> {
+ Dataset<Row> df =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.SPLIT_SIZE,
String.valueOf(optionSplitSize))
+ .load(tableName);
+ df.collect();
+ SparkPlan scan =
df.queryExecution().executedPlan().collectLeaves().head();
+
assertThat(scan.metrics().get("effectiveSplitTargetSize").get().value())
+ .isEqualTo(optionSplitSize);
+ });
+ }
+
+ @TestTemplate
+ void effectiveSplitTargetSizeWithoutSnapshot() {
Review Comment:
Applied in 513c06382, since the test now also covers pruned and deleted data.
##########
spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java:
##########
@@ -44,6 +61,180 @@ public class TestSparkReadMetrics extends
TestBaseWithCatalog {
@AfterEach
public void removeTables() {
sql("DROP TABLE IF EXISTS %s", tableName);
+ sql("DROP TABLE IF EXISTS dim");
+ }
+
+ @TestTemplate
+ void effectiveSplitTargetSize() throws NoSuchTableException, IOException {
+ long tableSplitSize = 128L * 1024 * 1024;
+ long optionSplitSize = 32L * 1024 * 1024;
+ int parallelism = 1000;
+ sql(
+ "CREATE TABLE %s (id BIGINT) USING iceberg TBLPROPERTIES
('read.split.target-size'='%s')",
+ tableName, tableSplitSize);
+ spark.range(100).coalesce(1).writeTo(tableName).append();
+
+ long scanSize = 0;
+ try (CloseableIterable<FileScanTask> tasks =
+ validationCatalog.loadTable(tableIdent).newScan().planFiles()) {
+ for (FileScanTask task : tasks) {
+ scanSize += task.sizeBytes();
+ }
+ }
+ long adaptiveSplitSize = TableScanUtil.adjustSplitSize(scanSize,
parallelism, tableSplitSize);
+ assertThat(adaptiveSplitSize).isNotEqualTo(tableSplitSize);
+
+ // adaptive sizing lowers the table target for a small scan
+ withSQLConf(
+ ImmutableMap.of(
+ SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_ENABLED,
+ "true",
+ SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_PARALLELISM,
+ String.valueOf(parallelism)),
+ () -> {
+ Dataset<Row> df = spark.sql(String.format("SELECT * FROM %s",
tableName));
+ df.collect();
+ SparkPlan scan =
df.queryExecution().executedPlan().collectLeaves().head();
+
assertThat(scan.metrics().get("effectiveSplitTargetSize").get().value())
+ .isEqualTo(adaptiveSplitSize);
+ });
+
+ // without adaptive sizing, the table property is the target
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_ENABLED,
"false"),
+ () -> {
+ Dataset<Row> df = spark.sql(String.format("SELECT * FROM %s",
tableName));
+ df.collect();
+ SparkPlan scan =
df.queryExecution().executedPlan().collectLeaves().head();
+
assertThat(scan.metrics().get("effectiveSplitTargetSize").get().value())
+ .isEqualTo(tableSplitSize);
+ });
+
+ // an explicit read option is used as is, even with adaptive sizing enabled
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_ENABLED,
"true"),
+ () -> {
+ Dataset<Row> df =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.SPLIT_SIZE,
String.valueOf(optionSplitSize))
+ .load(tableName);
+ df.collect();
+ SparkPlan scan =
df.queryExecution().executedPlan().collectLeaves().head();
+
assertThat(scan.metrics().get("effectiveSplitTargetSize").get().value())
+ .isEqualTo(optionSplitSize);
+ });
+ }
+
+ @TestTemplate
+ void effectiveSplitTargetSizeWithoutSnapshot() {
+ sql("CREATE TABLE %s (id BIGINT) USING iceberg", tableName);
+
+ // no task groups are planned for a table without a snapshot
Review Comment:
Applied in 513c06382.
##########
spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java:
##########
@@ -46,6 +52,80 @@ public void removeTables() {
sql("DROP TABLE IF EXISTS %s", tableName);
}
+ @TestTemplate
+ public void testEffectiveSplitTargetSize() throws NoSuchTableException {
+ sql(
+ "CREATE TABLE %s (id BIGINT) USING iceberg TBLPROPERTIES
('read.split.target-size'='134217728')",
+ tableName);
+ spark.range(100).coalesce(1).writeTo(tableName).append();
+
+ withSQLConf(
+ ImmutableMap.of(
+ SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_ENABLED,
+ "true",
+ SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_PARALLELISM,
+ "1000"),
+ () -> assertEffectiveSplitTargetSize(16L * 1024 * 1024));
+
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_ENABLED,
"false"),
+ () -> assertEffectiveSplitTargetSize(128L * 1024 * 1024));
+
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_ENABLED,
"true"),
+ () -> {
+ Dataset<Row> df =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.SPLIT_SIZE, String.valueOf(32L *
1024 * 1024))
+ .load(tableName);
+ assertEffectiveSplitTargetSize(df, 32L * 1024 * 1024);
+ });
+ }
+
+ @TestTemplate
+ public void testCustomMetricsAreInstantiableBySpark() {
Review Comment:
Fair point. In 513c06382 I removed `customMetricsAreLoadableBySpark` and
instead assert the value the SQL UI renders for this metric, the same way
#18751 does for `scanDuration` (`lastExecutedMetricValue(spark, new
EffectiveSplitTargetSize().description())`). That checks the user-facing
behavior and is limited to this PR's metric. It still catches the original
problem: with a metric class Spark cannot instantiate, the UI value is `N/A`
and the assertion fails.
##########
spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java:
##########
@@ -46,6 +52,80 @@ public void removeTables() {
sql("DROP TABLE IF EXISTS %s", tableName);
}
+ @TestTemplate
+ public void testEffectiveSplitTargetSize() throws NoSuchTableException {
Review Comment:
This follows the Iceberg AGENTS.md test guideline "Avoid using `test`
prefixes for newly added tests" (clarified in #16866), and @manuzhang asked for
it here. The most recent tests added to this file follow it as well: #18751
added `scanDurationAggregatedAcrossTasks` and
`scanDurationForNonVectorizedRead` without the prefix.
--
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]
