liurenjie1024 commented on code in PR #18709:
URL: https://github.com/apache/iceberg/pull/18709#discussion_r4509270509
##########
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:
I think we should move this after scanReport null check?
##########
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:
I prefer to drop this test, and merge it into testReadMetricsForV2Table,
testReadMetricsForV1Table
##########
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:
I don't think should be long. It should always have a value.
##########
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:
I think it would be better to add a runtime filter test.
##########
spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkStagedScan.java:
##########
@@ -108,6 +113,16 @@ public void testTaskSetPlanning() throws
NoSuchTableException, IOException {
assertThat(scanDF.javaRDD().getNumPartitions())
.as("Num partitions should match")
.isEqualTo(2);
+ scanDF.collect();
Review Comment:
Add a simple comment to explain what this does, as other parts.
##########
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:
I prefer to remove these simple helper functions to make it more readable.
##########
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:
We don't need this if the metrics has been set to init split size?
--
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]
