nastra commented on code in PR #14615:
URL: https://github.com/apache/iceberg/pull/14615#discussion_r2580186055
##########
spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java:
##########
@@ -267,6 +269,98 @@ public void testUnpartitionedTimestampFilter() {
"ts < cast('2017-12-22 00:00:00+00:00' as timestamp)"));
}
+ @TestTemplate
+ public void limitPushedDownToSparkScan() {
+ assumeThat(fileFormat)
+ .as("no need to run this across the entire test matrix")
+ .isEqualTo(FileFormat.PARQUET);
+
+ CaseInsensitiveStringMap options =
+ new CaseInsensitiveStringMap(ImmutableMap.of("path",
unpartitioned.toString()));
+
+ SparkScanBuilder builder =
+ new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
+
+ long limit = 23;
+ // simulate Spark pushing down the limit to the scan builder
+ builder.pushLimit((int) limit);
+ assertThat(builder).extracting("limit").isEqualTo((int) limit);
+
+ // verify batch scan
+ AbstractObjectAssert<?, ?> scanAssert =
assertThat(builder.build()).extracting("scan");
+ if (LOCAL == planningMode) {
+ scanAssert = scanAssert.extracting("scan");
+ }
+
+
scanAssert.extracting("context").extracting("minRowsRequested").isEqualTo(limit);
+
+ // verify changelog scan
+ assertThat(builder.buildChangelogScan())
+ .extracting("scan")
+ .extracting("context")
+ .extracting("minRowsRequested")
+ .isEqualTo(limit);
+
+ // verify CoW scan
+ assertThat(builder.buildCopyOnWriteScan())
+ .extracting("scan")
+ .extracting("scan")
+ .extracting("context")
+ .extracting("minRowsRequested")
+ .isEqualTo(limit);
+
+ // verify MoR scan
+ scanAssert = assertThat(builder.buildMergeOnReadScan()).extracting("scan");
+ if (LOCAL == planningMode) {
+ scanAssert = scanAssert.extracting("scan");
+ }
+
+
scanAssert.extracting("context").extracting("minRowsRequested").isEqualTo(limit);
+ }
+
+ @TestTemplate
+ public void limitPushedDownToSparkScanForMetadataTable() {
+ assumeThat(fileFormat)
+ .as("no need to run this across the entire test matrix")
+ .isEqualTo(FileFormat.PARQUET);
+
+ CaseInsensitiveStringMap options =
+ new CaseInsensitiveStringMap(ImmutableMap.of("path",
unpartitioned.toString()));
+
+ // load the snapshots metadata table
+ SparkScanBuilder builder =
+ new SparkScanBuilder(spark, TABLES.load(options.get("path") +
"#snapshots"), options);
+
+ long limit = 23;
+ // simulate Spark pushing down the limit to the scan builder
+ builder.pushLimit((int) limit);
+ assertThat(builder).extracting("limit").isEqualTo((int) limit);
+
+ // verify batch scan
+ assertThat(builder.build())
+ .extracting("scan")
+ .extracting("scan")
+ .extracting("context")
+ .extracting("minRowsRequested")
+ .isEqualTo(limit);
+
+ // verify CoW scan
+ assertThat(builder.buildCopyOnWriteScan())
Review Comment:
yeah distributed and local planning is both already handled by the test
parameterization. For the metadata tables I just wanted to make sure that the
limit is properly pushed down to the `TableScanContext` independent of whether
buildCopyOnWriteScan/buildMergeOnReadScan actually makes sense on a metadata
table or not, since technically these are callable from a user's perspective
--
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]
