gerashegalov commented on code in PR #18709:
URL: https://github.com/apache/iceberg/pull/18709#discussion_r4513567816
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java:
##########
@@ -114,6 +116,7 @@ abstract class SparkScan implements Scan,
SupportsReportStatistics {
// lazy variables
private StructType readSchema;
+ private Long effectiveSplitTargetSize;
Review Comment:
Done in c5ebd6fa7: it's now a `long` initialized to the configured split
size and overwritten by adaptive sizing.
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java:
##########
@@ -301,12 +304,15 @@ public String toString() {
@Override
public CustomTaskMetric[] reportDriverMetrics() {
ScanReport scanReport = scanReportSupplier != null ?
scanReportSupplier.get() : null;
+ List<CustomTaskMetric> driverMetrics = Lists.newArrayList();
- if (scanReport == null) {
- return new CustomTaskMetric[0];
+ if (effectiveSplitTargetSize != null) {
Review Comment:
`SparkStagedScan` passes a null scan-report supplier, so moving this after
the null check would drop the metric for staged rewrite scans (covered in
`TestSparkStagedScan`). The split target doesn't depend on the scan report, so
I kept it unconditional.
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkStagedScan.java:
##########
@@ -66,6 +66,7 @@ protected List<ScanTaskGroup<ScanTask>> taskGroups() {
taskSetId);
this.taskGroups = TableScanUtil.planTaskGroups(tasks, splitSize,
splitLookback, openFileCost);
+ recordEffectiveSplitTargetSize(splitSize);
Review Comment:
Done in c5ebd6fa7: removed, since the initial value is the staged scan's
split size.
##########
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:
Done in c5ebd6fa7: folded the check into `testReadMetricsForV1Table` and
`testReadMetricsForV2Table`.
##########
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() {
+ sql("CREATE TABLE %s (id BIGINT) USING iceberg", tableName);
+
+ Dataset<Row> df = spark.sql(String.format("SELECT * FROM %s", tableName));
+ SparkPlan scanPlan =
df.queryExecution().executedPlan().collectLeaves().head();
+
+ assertThat(scanPlan).isInstanceOf(BatchScanExec.class);
+ CustomMetric[] metrics = ((BatchScanExec)
scanPlan).scan().supportedCustomMetrics();
+
+ // Spark's SQLAppStatusListener creates each metric reflectively by class
name using a public
+ // zero-argument constructor; a private or nested class silently drops the
metric from the UI
+ assertThat(metrics).isNotEmpty();
+ for (CustomMetric metric : metrics) {
+ scala.collection.immutable.Seq<CustomMetric> loaded =
+ Utils.loadExtensions(
+ CustomMetric.class,
+
JavaConverters.asScalaBuffer(List.of(metric.getClass().getName())).toSeq(),
+ spark.sparkContext().conf());
+ assertThat(loaded.size()).as("metric %s", metric.name()).isEqualTo(1);
+ }
+ }
+
+ private void assertEffectiveSplitTargetSize(long expectedSize) {
+ Dataset<Row> df = spark.sql(String.format("SELECT * FROM %s", tableName));
+ assertEffectiveSplitTargetSize(df, expectedSize);
+ }
+
+ private void assertEffectiveSplitTargetSize(Dataset<Row> df, long
expectedSize) {
Review Comment:
Done in c5ebd6fa7.
##########
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:
Added `effectiveSplitTargetSizeWithRuntimeFilter` in c5ebd6fa7. Note that
the scan-report metrics (e.g. `resultDataFiles`) reflect planning before the
runtime filter, while the split target is recorded when task groups are
re-planned after it.
--
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]
