voonhous opened a new issue, #20534:
URL: https://github.com/apache/hudi/issues/20534
### Describe the problem
With data skipping on, a query filtering a column-stats-indexed column on a
large `IN` list fails on the driver with `StackOverflowError` or
`OutOfMemoryError`. The same query with data skipping off runs in about 100 ms.
Spark rewrites an `IN` list of more than 10 literals into a flat `InSet`.
`Verdict` (`dataskipping/Verdict.scala:42-44`) expands it back with
`.reduce(Or)`: a left-deep `OR` with one level per literal.
`SparkBaseIndexSupport.getCandidateFiles` then applies it with
`indexDf.where(...)`, and Spark's analyzer and codegen recurse over that tree.
Codegen also gets quadratic: every `OR` level re-copies its child's generated
code.
`fallback` mode does not help. Scala's `NonFatal` does not match
`StackOverflowError` or `OutOfMemoryError`, so the query fails in both modes.
Partition stats and expression index pruning use the same translation.
### Repro
COW table, 4 base files, column stats on a `bigint` column,
`hoodie.enable.data.skipping=true`, `WHERE v IN (0, 1, ..., n-1)`. Spark 3.5,
JDK 11, `-Xss1m` (the Linux x64 default), `-Xmx4g`, master `9c75d0b4`:
| n | data skipping on (`fallback` and `strict`) | data skipping off |
| --- | --- | --- |
| 500 | OK, 2.5 s | 0.1 s |
| 1000 | OK, 8.7 s | 0.06 s |
| 2000 | `OutOfMemoryError: GC overhead limit exceeded` after 36-45 s | 0.08
s |
| 4000, 8000 | `StackOverflowError` (analyzer) | 0.1-0.2 s |
When the column stats table is small enough to become a `LocalRelation`, the
overflow moves into `Or.doGenCode` under `ConvertToLocalRelation` (seen at n =
2000). The exact threshold depends on stack size and JIT state.
<details>
<summary>StackOverflowError, n = 4000 (27,584 frames, trimmed)</summary>
```
java.lang.StackOverflowError
at
org.apache.spark.sql.catalyst.trees.BinaryLike.mapChildren(TreeNode.scala:1242)
at
org.apache.spark.sql.catalyst.trees.BinaryLike.mapChildren$(TreeNode.scala:1241)
at
org.apache.spark.sql.catalyst.expressions.BinaryExpression.mapChildren(Expression.scala:653)
at
org.apache.spark.sql.catalyst.analysis.ColumnResolutionHelper.$anonfun$resolveExpression$1(ColumnResolutionHelper.scala:187)
at
org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:76)
at
org.apache.spark.sql.catalyst.analysis.ColumnResolutionHelper.innerResolve$1(ColumnResolutionHelper.scala:135)
at
org.apache.spark.sql.catalyst.analysis.ColumnResolutionHelper.$anonfun$resolveExpression$9(ColumnResolutionHelper.scala:187)
... (the 7 frames above repeat about 3,900 times, once per OR level)
at
org.apache.spark.sql.execution.QueryExecution.analyzed(QueryExecution.scala:74)
at
org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:66)
at org.apache.spark.sql.Dataset.<init>(Dataset.scala:206)
at org.apache.spark.sql.Dataset$.apply(Dataset.scala:76)
at org.apache.spark.sql.Dataset.filter(Dataset.scala:1676)
at org.apache.spark.sql.Dataset.where(Dataset.scala:1704)
at
org.apache.hudi.SparkBaseIndexSupport.getCandidateFiles(SparkBaseIndexSupport.scala:123)
at
org.apache.hudi.ColumnStatsIndexSupport.$anonfun$computeCandidateFileNames$2(ColumnStatsIndexSupport.scala:102)
at
org.apache.hudi.ColumnStatsIndexSupport.$anonfun$loadTransposed$2(ColumnStatsIndexSupport.scala:172)
at
org.apache.hudi.HoodieCatalystUtils$.withPersistedDataset(HoodieCatalystUtils.scala:42)
at
org.apache.hudi.ColumnStatsIndexSupport.loadTransposed(ColumnStatsIndexSupport.scala:149)
at
org.apache.hudi.ColumnStatsIndexSupport.computeCandidateFileNames(ColumnStatsIndexSupport.scala:101)
at
org.apache.hudi.SparkBaseIndexSupport.computeCandidateIsStrict(SparkBaseIndexSupport.scala:67)
at
org.apache.hudi.HoodieFileIndex.lookupCandidateFilesInMetadataTable(HoodieFileIndex.scala:428)
at
org.apache.hudi.HoodieFileIndex.filterFileSlices(HoodieFileIndex.scala:278)
at org.apache.hudi.HoodieFileIndex.listFiles(HoodieFileIndex.scala:205)
```
</details>
<details>
<summary>OutOfMemoryError, n = 2000, whole-stage codegen of the column stats
filter (trimmed)</summary>
```
java.lang.OutOfMemoryError: GC overhead limit exceeded
at java.base/java.util.Arrays.copyOfRange(Arrays.java:4030)
at java.base/java.lang.StringLatin1.newString(StringLatin1.java:715)
at java.base/java.lang.String.substring(String.java:1879)
at
scala.collection.immutable.StringLike.stripMargin(StringLike.scala:193)
at
org.apache.spark.sql.catalyst.expressions.codegen.Block.toString(javaCode.scala:143)
at
org.apache.spark.sql.catalyst.expressions.codegen.CodeBlock.code$lzycompute(javaCode.scala:296)
at
org.apache.spark.sql.catalyst.expressions.codegen.Block.toString(javaCode.scala:143)
at
org.apache.spark.sql.catalyst.expressions.codegen.CodeBlock.code$lzycompute(javaCode.scala:296)
at
org.apache.spark.sql.catalyst.expressions.codegen.Block.toString(javaCode.scala:143)
at
org.apache.spark.sql.catalyst.expressions.codegen.Block.length(javaCode.scala:150)
at
org.apache.spark.sql.catalyst.expressions.Expression.reduceCodeSize(Expression.scala:215)
... (the JVM's preallocated OOM keeps only 32 frames)
```
</details>
<details>
<summary>StackOverflowError, n = 2000, column stats as a LocalRelation
(trimmed)</summary>
```
java.lang.StackOverflowError
at
org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext.freshName(CodeGenerator.scala:594)
at
org.apache.spark.sql.catalyst.expressions.Expression.$anonfun$genCode$3(Expression.scala:197)
at scala.Option.getOrElse(Option.scala:189)
at
org.apache.spark.sql.catalyst.expressions.Expression.genCode(Expression.scala:196)
at
org.apache.spark.sql.catalyst.expressions.Or.doGenCode(predicates.scala:913)
... (Or.doGenCode repeats 1,999 times)
at
org.apache.spark.sql.catalyst.expressions.codegen.GeneratePredicate$.create(GeneratePredicate.scala:41)
at
org.apache.spark.sql.catalyst.expressions.Predicate$.createCodeGeneratedObject(predicates.scala:73)
at
org.apache.spark.sql.catalyst.expressions.CodeGeneratorWithInterpretedFallback.createObject(CodeGeneratorWithInterpretedFallback.scala:50)
at
org.apache.spark.sql.catalyst.expressions.Predicate$.create(predicates.scala:86)
at
org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation$$anonfun$apply$47.applyOrElse(Optimizer.scala:2189)
```
</details>
### Suggested fix
Build a balanced `OR` (log2(n) deep) with Spark's
`PredicateHelper.buildBalancedPredicate`, which exists in every Spark line Hudi
supports (3.3 to 4.2). Pruning stays exactly the same.
--
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]
