rich7420 commented on code in PR #6847:
URL: https://github.com/apache/datafusion-comet/pull/6847#discussion_r4344990621
##########
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:
Fixed with `withRecordCount(rows.size.toLong)`. The Spark 3.5 / Scala 2.12
`-Pstrict-warnings` test-compile now passes locally.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -733,36 +740,42 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
val left = icebergExprToProto(
IcebergReflection.getMethod(exprClass, "left").invoke(icebergExpr),
output,
- pageIndexUnsupportedColumns)
+ pageIndexUnsupportedColumns,
+ allowPartial)
val right = icebergExprToProto(
IcebergReflection.getMethod(exprClass, "right").invoke(icebergExpr),
output,
- pageIndexUnsupportedColumns)
+ pageIndexUnsupportedColumns,
+ allowPartial)
(left, right) match {
- // Push the residual only if it converts whole. Dropping a conjunct is
safe in positive
- // position but strengthens the predicate under a NOT (De Morgan),
which would wrongly
- // prune, and tracking polarity across arbitrary nesting is error
prone. So an
- // unconvertible conjunct elides the whole residual; the post-scan
CometFilter is exact.
case (Some(l), Some(r)) => Some(logicalPredicate(isAnd = true, l, r))
+ // Dropping a positive conjunct weakens the predicate; the post-scan
filter stays exact.
+ // Inside NOT, every descendant must convert exactly, including nested
AND/OR nodes.
+ case (l, r) if allowPartial => l.orElse(r)
Review Comment:
Fixed. Equality on a nullable column is no longer pushed when it appears
anywhere under `NOT`, so `l IS NOT NULL AND NOT (id <=> 1)` keeps the row where
`id` is NULL. The runtime regression test covers your expression and its nested
variants. This fix now lives in #7156.
##########
docs/source/user-guide/latest/iceberg.md:
##########
@@ -113,8 +113,30 @@ The native Iceberg reader supports the following features:
NULL checks on struct, array, and map columns still use native scans and
return correct
results, but are not pushed into iceberg-rust, which binds accessors only for
primitive fields.
-These residuals provide no native row-group pruning, nor do conjunctions
containing them;
-safe partial pruning is tracked in
[#5883](https://github.com/apache/datafusion-comet/issues/5883).
+These null checks provide no native row-group pruning. In a positive
conjunction, supported
+primitive predicates still prune row groups and pages, while the post-scan
filter evaluates the
+complete condition. `OR` requires a safely pushable predicate from both
branches, and expressions
+inside `NOT` must convert completely before they can be pushed. When only part
of a condition can
+be pushed, predicates on `BIGINT` and `DOUBLE` columns in it are pushed only
for `=` and null
+checks, because a data file can store a narrower type than its column without
any record of it.
+
+Tables with a `schema.name-mapping.default` property push a conjunction only
when every predicate
+in it can be pushed. The `add_files`, `migrate` and `snapshot` procedures set
this property when
+they import data files, and an imported file can store a narrower type than
its column.
+
+Spark treats `-0.0` as equal to `0.0` and sorts every NaN above all numbers,
but iceberg-rust
+uses IEEE 754 total ordering. On `FLOAT` and `DOUBLE` columns, range
comparisons and comparisons
+with `0.0`, `-0.0` or NaN are therefore evaluated after the native scan.
+
+Some other predicates are also evaluated only after the native scan:
+
+- Negated equality on a nullable column. Iceberg passes `<=>` as `=`, so for
`NOT (a <=> 1)` the
+ native reader would drop NULL rows that Spark keeps, and `a <> 1` reaches
Comet in the same
+ form.
+- Comparisons other than `=` with a literal on a column whose type was
promoted, for example
+ from `INT` to `BIGINT`, because older data files keep the original type. If
expiring
+ snapshots removed older schemas, this applies to every `BIGINT` and `DOUBLE`
column. A
+ promotion made before Iceberg 0.12, which kept no schema history, cannot be
detected.
Review Comment:
Applied your version, thanks.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -679,17 +679,66 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
* - An exception means reflection over Iceberg's expression API did
something unexpected, so
* no such decision was reached. It propagates instead of folding into
`None`, which would
* make the two indistinguishable; see [[serializeResidual]].
+ *
+ * @param promotedColumns
+ * columns whose older data files may hold a narrower type; see
+ * [[IcebergReflection.promotedColumns]]
+ * @param partialPushdown
+ * whether a residual that does not convert whole may push the conjuncts
that do, leaving the
+ * rest to the post-scan filter
*/
def icebergExprToProto(
icebergExpr: Any,
output: Seq[Attribute],
- pageIndexUnsupportedColumns: Set[String]):
Option[OperatorOuterClass.IcebergPredicate] = {
+ pageIndexUnsupportedColumns: Set[String],
+ promotedColumns: Set[String] = Set.empty,
+ partialPushdown: Boolean = true):
Option[OperatorOuterClass.IcebergPredicate] = {
Review Comment:
Both are now required in `icebergExprToProto` and `serializeResidual`, and
the tests pass them explicitly.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -679,17 +679,66 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
* - An exception means reflection over Iceberg's expression API did
something unexpected, so
* no such decision was reached. It propagates instead of folding into
`None`, which would
* make the two indistinguishable; see [[serializeResidual]].
+ *
+ * @param promotedColumns
+ * columns whose older data files may hold a narrower type; see
+ * [[IcebergReflection.promotedColumns]]
+ * @param partialPushdown
+ * whether a residual that does not convert whole may push the conjuncts
that do, leaving the
+ * rest to the post-scan filter
*/
def icebergExprToProto(
icebergExpr: Any,
output: Seq[Attribute],
- pageIndexUnsupportedColumns: Set[String]):
Option[OperatorOuterClass.IcebergPredicate] = {
+ pageIndexUnsupportedColumns: Set[String],
+ promotedColumns: Set[String] = Set.empty,
+ partialPushdown: Boolean = true):
Option[OperatorOuterClass.IcebergPredicate] = {
+ val attributes = output.map(attr => attr.name -> attr).toMap
+ def convert(promoted: Set[String], partial: Boolean) =
+ icebergExprToProto(
+ icebergExpr,
+ ResidualScope(attributes, pageIndexUnsupportedColumns, promoted,
partial),
+ underNot = false)
+ // Pushing part of a residual must not rely on the schema history, which
misses some
+ // promotions, such as those made before Iceberg 0.12. So when the
residual does not convert
+ // whole, every BIGINT and DOUBLE column counts as promoted. Whole
residuals rely on the
+ // history, since limiting them too would stop range pruning on all such
columns until
+ // apache/iceberg-rust#3122 is fixed.
+ convert(promotedColumns, partial = false).orElse {
+ if (partialPushdown) {
+ val widened = output.collect {
+ case attr if attr.dataType == LongType || attr.dataType ==
DoubleType => attr.name
+ }
+ convert(promotedColumns ++ widened, partial = true)
+ } else {
+ None
+ }
+ }
+ }
+
+ /** What [[icebergExprToProto]] consults at every node of a residual. */
+ private case class ResidualScope(
+ attributes: Map[String, Attribute],
+ pageIndexUnsupportedColumns: Set[String],
+ promotedColumns: Set[String],
+ partialPushdown: Boolean)
Review Comment:
Removed it. The recursive walk now takes plain parameters instead, and its
Scaladoc documents `attributes`, `promotedColumns`, `dropUnconvertedConjuncts`
and `underNot`. `partialPushdown` is the table-level switch, and it sets
`dropUnconvertedConjuncts` for a single pass over the residual.
--
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]
