sunchao commented on code in PR #6847:
URL: https://github.com/apache/datafusion-comet/pull/6847#discussion_r4329119726
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -4201,6 +4203,440 @@ class CometIcebergNativeSuite
}
}
+ test("partial residual pushdown preserves complex filters and reduces reader
work") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withHadoopCatalog("test_cat") {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+ val table = "test_cat.db.partial_residual_test"
+ // Keep split sizing fixed so repeated footer reads do not dominate
the byte comparison.
+ spark.sql(s"""
+ CREATE TABLE $table (
+ id INT,
+ l ARRAY<INT>,
+ m MAP<STRING, INT>,
+ s STRUCT<v: INT>,
+ payload STRING
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'read.split.adaptive-size.enabled' = 'false',
+ 'write.parquet.row-group-size-bytes' = '16384',
+ 'write.parquet.compression-codec' = 'uncompressed'
+ )
+ """)
+ try {
+ spark
+ .range(10000)
+ .repartition(1)
+ .sortWithinPartitions("id")
+ .selectExpr(
+ "CASE WHEN id % 97 = 0 THEN NULL ELSE CAST(id AS INT) END AS id",
+ "CASE WHEN id % 2 = 0 THEN array(CAST(id AS INT)) ELSE NULL END
AS l",
+ "CASE WHEN id % 3 = 0 THEN map('v', CAST(id AS INT)) ELSE NULL
END AS m",
+ "CASE WHEN id % 5 = 0 THEN named_struct('v', CAST(id AS INT))
ELSE NULL END AS s",
+ "repeat(sha2(CAST(id AS STRING), 256), 2) AS payload")
+ .write
+ .format("iceberg")
+ .mode("append")
+ .saveAsTable(table)
+
+ // One unpartitioned file containing several row groups: the byte
savings must come
+ // from the reader, rather than Iceberg's manifest/file pruning or
partition selection.
+ val files = spark.sql(s"SELECT file_path FROM
$table.files").collect()
+ assert(files.length == 1, s"Expected one data file, got
${files.length}")
+ val reader = org.apache.parquet.hadoop.ParquetFileReader.open(
+ org.apache.parquet.hadoop.util.HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(files.head.getString(0)),
+ spark.sessionState.newHadoopConf()))
+ try {
+ assert(reader.getRowGroups.size > 3, "Fixture needs multiple row
groups for pruning")
+ } finally {
+ reader.close()
+ }
+
+ def checkAndGetScan(predicate: String): CometIcebergNativeScanExec =
{
+ val query = s"SELECT * FROM $table WHERE $predicate"
+ val (_, plan) = checkSparkAnswer(query)
+ val scans = collectIcebergNativeScans(plan)
+ assert(scans.length == 1, s"Expected one native scan for
$query:\n$plan")
+ scans.head
+ }
+
+ val fullScan = checkAndGetScan("l IS NOT NULL")
+ val fullBytes = fullScan.metrics("bytes_scanned").value
+ assert(fullBytes > 0, "Full read must report bytes")
+ assert(fullScan.metrics("output_rows").value == 10000L)
+
+ val partialScan = checkAndGetScan("l IS NOT NULL AND id BETWEEN 1000
AND 1099")
+ val partialBytes = partialScan.metrics("bytes_scanned").value
+ assert(fullScan.metrics("num_splits").value > 0L)
+ assert(partialScan.metrics("num_splits").value ==
fullScan.metrics("num_splits").value)
+ assert(
+ partialScan.metrics("output_rows").value < 1000L,
+ "The native reader must apply the primitive residual before the
complex filter")
+ assert(
+ partialBytes * 2 < fullBytes,
+ s"Partial residual must prune reads: partial=$partialBytes,
full=$fullBytes")
+
+ Seq(
+ "m IS NOT NULL AND id BETWEEN 1000 AND 1099",
+ "s IS NULL AND id BETWEEN 1000 AND 1099",
+ "id >= 1000 AND (l IS NOT NULL OR m IS NULL)",
+ "(id < 100 AND l IS NOT NULL) OR (id >= 9900 AND m IS NOT NULL)",
+ "id < 100 OR l IS NOT NULL",
+ "NOT (id < 2000 AND l IS NOT NULL)",
+ "NOT ((id < 100 AND l IS NOT NULL) OR (id >= 9900 AND m IS NOT
NULL))")
+ .foreach { predicate =>
+ val _ = checkAndGetScan(predicate)
+ }
+ } finally {
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+ }
+
+ test("floating-point residuals preserve signed zeros in native scans") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withHadoopCatalog("test_cat") {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+ val table = "test_cat.db.floating_residual_test"
+ spark.sql(s"CREATE TABLE $table (id INT, f FLOAT, d DOUBLE, l
ARRAY<INT>) USING iceberg")
+ try {
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ // One file containing both zeros keeps file pruning from deciding
the comparison.
+ // Distinct ids make a missing signed-zero row observable without
comparing floats.
+ spark
+ .sql("""
+ SELECT id, CAST(v AS FLOAT) AS f, CAST(v AS DOUBLE) AS d, l
+ FROM VALUES
+ (1, '-0.0', array(1)),
+ (2, '0.0', array(1)),
+ (3, '1.0', CAST(NULL AS ARRAY<INT>)),
+ (4, '-1.0', array(1)),
+ (5, NULL, array(1)),
+ (6, 'NaN', array(1))
+ AS input(id, v, l)
+ """)
+ .coalesce(1)
+ .write
+ .format("iceberg")
+ .mode("append")
+ .saveAsTable(table)
+ assert(spark.sql(s"SELECT file_path FROM
$table.files").collect().length == 1)
+ val rows = spark.sql(s"SELECT id, f, d FROM
$table").collect().sortBy(_.getInt(0))
+ assert(java.lang.Float.floatToRawIntBits(rows(0).getFloat(1)) ==
Int.MinValue)
+ assert(java.lang.Double.doubleToRawLongBits(rows(0).getDouble(2))
== Long.MinValue)
+ assert(java.lang.Float.floatToRawIntBits(rows(1).getFloat(1)) == 0)
+ assert(java.lang.Double.doubleToRawLongBits(rows(1).getDouble(2))
== 0L)
+ }
+
+ for ((column, dataType) <- Seq("f" -> "FLOAT", "d" -> "DOUBLE")) {
+ val zero = s"CAST(0 AS $dataType)"
+ val negativeZero = s"CAST('-0.0' AS $dataType)"
+ val one = s"CAST(1 AS $dataType)"
+ val minusOne = s"CAST(-1 AS $dataType)"
+ val predicates = Seq(
+ s"$column = $zero",
+ s"$column = $negativeZero",
+ s"$column >= $zero",
+ s"$column <= $negativeZero",
+ s"$column IN ($zero, $one)",
+ s"NOT ($column < $zero)",
+ // Nonzero literals push equality and IN; range comparisons stay
out.
+ s"$column = $one",
+ s"$column IN ($one, $minusOne)",
+ s"NOT ($column IN ($one, $minusOne) AND id > 1)",
+ s"$column < $one",
+ s"$column > $minusOne",
+ s"NOT ($column = $one)",
+ s"$column = $zero OR id = 3",
+ s"NOT ($column = $zero AND id > 1)",
+ s"$column IS NULL",
+ s"$column IS NOT NULL")
+ for (predicate <- predicates;
+ prefix <- Seq("", "l IS NOT NULL AND ")) {
+ checkIcebergNativeScan(s"SELECT id FROM $table WHERE
$prefix($predicate)")
+ }
+ }
+ } finally {
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+ }
+
+ test("negated null-safe equality keeps NULL rows in native scans") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withHadoopCatalog("test_cat") {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+ val table = "test_cat.db.null_safe_residual_test"
+ spark.sql(s"CREATE TABLE $table (id INT, b BOOLEAN, l ARRAY<INT>)
USING iceberg")
+ try {
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ // One file keeps file pruning from deciding the comparison.
+ spark
+ .sql("""
+ SELECT * FROM VALUES
+ (NULL, NULL, array(1)), (1, true, array(1)), (2, false,
array(1))
+ AS t(id, b, l)
+ """)
+ .coalesce(1)
+ .write
+ .format("iceberg")
+ .mode("append")
+ .saveAsTable(table)
+ assert(spark.sql(s"SELECT file_path FROM
$table.files").collect().length == 1)
+ }
+ // Iceberg converts `id <=> 1` to an equality, which is false for a
NULL id in Spark but
+ // NULL in the native reader, so only its negation can drop the NULL
row. Spark also
+ // writes that negation as IS DISTINCT FROM and, for booleans, IS
NOT TRUE.
+ val predicates = Seq(
+ "NOT (id <=> 1)",
+ "NOT (id <=> 1 AND id < 5)",
+ "NOT (id <=> 1) OR id = 2",
+ "id IS DISTINCT FROM 1",
+ "b IS NOT TRUE",
+ "NOT b",
+ "NOT (id = 1)",
+ "id != 1")
+ for (predicate <- predicates;
+ prefix <- Seq("", "l IS NOT NULL AND ")) {
+ checkIcebergNativeScan(s"SELECT id FROM $table WHERE
$prefix($predicate)")
+ }
+ } finally {
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+ }
+
+ test("literal residuals on promoted columns keep rows from older files") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withHadoopCatalog("test_cat") {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+ val table = "test_cat.db.promoted_residual_test"
+ spark.sql(s"CREATE TABLE $table (id INT, f FLOAT, l ARRAY<INT>) USING
iceberg")
+ try {
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ // One file, so that its FLOAT bounds keep it for f = 0.1D below.
+ spark.sql(s"""INSERT INTO $table SELECT /*+ COALESCE(1) */ * FROM
VALUES
+ (1, 0.5, array(1)), (2, 2.5, array(1)), (3, 0.1, array(1)), (6,
0.05, array(1))
+ AS v(id, f, l)""")
+ spark.sql(s"ALTER TABLE $table ALTER COLUMN id TYPE BIGINT")
+ spark.sql(s"ALTER TABLE $table ALTER COLUMN f TYPE DOUBLE")
+ spark.sql(s"INSERT INTO $table VALUES (4, 0.5D, array(1)), (5,
4.5D, array(1))")
+ }
+ // The first file still stores INT and FLOAT. iceberg-rust skips its
pages for range and
+ // IN predicates on the promoted types, and casts 3000000000 to a
NULL INT.
+ val predicates = Seq(
+ "id > 0",
+ "id < 3",
+ "id IN (1, 2)",
+ "id < 3000000000",
+ "id = 2",
+ "id = 2 OR id = 3000000000",
+ "f IN (0.5D, 4.5D)",
+ "f = 0.5D",
+ "f = 0.1D")
+ def checkPredicates(): Unit =
+ for (predicate <- predicates;
+ prefix <- Seq("", "l IS NOT NULL AND ")) {
+ checkIcebergNativeScan(s"SELECT id FROM $table WHERE
$prefix($predicate)")
+ }
+ checkPredicates()
+
+ // The native reader narrows 0.1D to the FLOAT 0.1, which Spark does
not consider equal
+ // to it. The first file's bounds, 0.05 to 2.5, keep it in the scan
for that.
+ val narrowed = s"SELECT id FROM $table WHERE f = 0.1D"
+ val df = spark.sql(narrowed)
+ df.collect()
+ val scanned =
+ collectIcebergNativeScans(df.queryExecution.executedPlan).head
+ .metrics("resultDataFiles")
+ .value
+ assert(scanned == 1, s"Expected the first file to be scanned,
scanned $scanned files")
+
+ // Expiring the first snapshot with cleanExpiredMetadata (Iceberg
1.8+) drops the original
+ // schema, while the file that snapshot wrote stays in the table.
+ if (icebergVersionAtLeast(1, 8)) {
+ val icebergTable =
+ loadIcebergTable(spark, "test_cat", "db",
"promoted_residual_test")
+ .asInstanceOf[org.apache.iceberg.Table]
+ val expire = icebergTable
+ .expireSnapshots()
+ .expireOlderThan(System.currentTimeMillis())
+ .retainLast(1)
+ classOf[org.apache.iceberg.ExpireSnapshots]
+ .getMethod("cleanExpiredMetadata", classOf[Boolean])
+ .invoke(expire, java.lang.Boolean.TRUE)
+ expire.commit()
+ spark.sql(s"REFRESH TABLE $table")
+ val schemas = loadIcebergTable(spark, "test_cat", "db",
"promoted_residual_test")
+ .asInstanceOf[org.apache.iceberg.Table]
+ .schemas()
+ assert(schemas.size == 1 && !schemas.containsKey(0))
+ assert(spark.sql(s"SELECT * FROM $table").count() == 6)
+ checkPredicates()
+ }
+ } finally {
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+ }
+
+ test("residuals on tables with imported files are pushed only whole") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withHadoopCatalog("test_cat") {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+ val table = "test_cat.db.imported_residual_test"
+ spark.sql(s"CREATE TABLE $table (id BIGINT, t INT, f FLOAT) USING
iceberg")
+ try {
+ withTempDir { dir =>
+ // The file stores id as INT and t as TINYINT. Importing it the
way add_files does
+ // adds a name mapping but records no schema change.
+ val source = new File(dir, "source").getAbsolutePath
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .sql("SELECT CAST(id AS INT) AS id, CAST(id AS TINYINT) AS t,
" +
+ "CAST(id AS FLOAT) AS f FROM range(1, 4)")
+ .coalesce(1)
+ .write
+ .parquet(source)
+ }
+ val icebergTable = loadIcebergTable(spark, "test_cat", "db",
"imported_residual_test")
+ .asInstanceOf[org.apache.iceberg.Table]
+ icebergTable
+ .updateProperties()
+ .set(
+ TableProperties.DEFAULT_NAME_MAPPING,
+
NameMappingParser.toJson(MappingUtil.create(icebergTable.schema())))
+ .commit()
+ val file = new
File(source).listFiles().find(_.getName.endsWith(".parquet")).get
+ icebergTable
+ .newAppend()
+ .appendFile(
+ DataFiles
+ .builder(icebergTable.spec())
+ .withPath(file.getAbsolutePath)
+ .withFileSizeInBytes(file.length())
+ .withRecordCount(3)
+ .withFormat(FileFormat.PARQUET)
+ .build())
+ .commit()
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark.sql(s"INSERT INTO $table VALUES (4, 4, 4.0)")
+ }
+ spark.sql(s"REFRESH TABLE $table")
+
+ // The float range is never pushed, so each residual converts only
in part. Pushing
+ // that part would skip the imported file's pages, or narrow 200
to a NULL TINYINT.
+ for (predicate <- Seq(
+ "id > 0",
+ "id < 3000000000",
+ "id IN (1, 2)",
+ "t < 200",
+ "t IN (1, 200)")) {
+ checkIcebergNativeScan(
+ s"SELECT id FROM $table WHERE f > CAST(0.5 AS FLOAT) AND
$predicate")
+ }
+ checkIcebergNativeScan(s"SELECT id FROM $table WHERE id = 2 AND t
= 2")
+ }
+ } finally {
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+ }
+
+ test("partial residuals push only equality on BIGINT and DOUBLE columns") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withHadoopCatalog("test_cat") {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+ val table = "test_cat.db.unrecorded_promotion_test"
+ spark.sql(s"CREATE TABLE $table (id BIGINT, f FLOAT, d DOUBLE) USING
iceberg")
+ try {
+ withTempDir { dir =>
+ // The file stores id as INT and d as FLOAT under the table's
field ids, and neither
+ // a schema nor a name mapping records it, as after a promotion
before Iceberg 0.12.
+ val source = new File(dir, "source").getAbsolutePath
+ def fieldId(id: Long) = new
MetadataBuilder().putLong("parquet.field.id", id).build()
+ val schema = new StructType()
+ .add("id", IntegerType, nullable = true, fieldId(1))
+ .add("f", FloatType, nullable = true, fieldId(2))
+ .add("d", FloatType, nullable = true, fieldId(3))
+ val rows = Seq(Row(1, 1.0f, 0.5f), Row(2, 2.0f, 2.5f), Row(3,
3.0f, 4.5f))
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ "spark.sql.parquet.fieldId.write.enabled" -> "true") {
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(rows, 1),
schema)
+ .write
+ .parquet(source)
+ }
+ val icebergTable =
+ loadIcebergTable(spark, "test_cat", "db",
"unrecorded_promotion_test")
+ .asInstanceOf[org.apache.iceberg.Table]
+ val file = new
File(source).listFiles().find(_.getName.endsWith(".parquet")).get
+ icebergTable
+ .newAppend()
+ .appendFile(
+ DataFiles
+ .builder(icebergTable.spec())
+ .withPath(file.getAbsolutePath)
+ .withFileSizeInBytes(file.length())
+ .withRecordCount(rows.size)
Review Comment:
[P2] Convert `rows.size` to `Long` explicitly before passing it to
`withRecordCount`. The required Spark 3.5/Scala 2.12 strict build enables
numeric-widening warnings as errors, so this new fixture fails test compilation
with `implicit numeric widening` instead of completing successfully. This
blocks the required CI check. Using `withRecordCount(rows.size.toLong)`
preserves the fixture's behavior and removes the compilation failure.
Evidence: Exact-head CI job
https://github.com/apache/datafusion-comet/actions/runs/39475272780/job/118816221478
ran `./mvnw -B test-compile -Pspark-3.5,delta -Pstrict-warnings -DskipTests`
and reported `CometIcebergNativeSuite.scala:4612: implicit numeric widening`,
followed by failure of `scala-test-compile-first`. The changed line passes the
Int-valued `rows.size` to the Long-valued record-count API.
--
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]
