You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "comphead (via GitHub)" <gi...@apache.org> on 2026/03/27 21:06:29 UTC
[I] `native_datafusion` doesn't use all available parallelism for scan [datafusion-comet]
comphead opened a new issue, #3817:
URL: https://github.com/apache/datafusion-comet/issues/3817
### What is the problem the feature request solves?
Observed the issue when Comet is not fully utilizing Spark cluster parallelism.
Input: 1200 HDFS files, number of Spark planned tasks: 1800. Every file is splittable, so Spark utilizes all 1800 scanning and writing the shuffle whereas Comet utilizing only 1200 tasks having 600 idle.
I was not able to reproduce the same locally, will try on local HDFS later
### Describe the potential solution
_No response_
### Additional context
_No response_
--
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: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] `native_datafusion` doesn't use all available parallelism for scan [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on issue #3817:
URL: https://github.com/apache/datafusion-comet/issues/3817#issuecomment-4158621246
Interesting observation, if I run a query with just count
```
spark.read.parquet("location").count()
```
Then all 1800 tasks reads from parquet
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] `native_datafusion` doesn't use all available parallelism for scan [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #3817:
URL: https://github.com/apache/datafusion-comet/issues/3817#issuecomment-4246263540
fwiw, Claude analysis of options to fix:
Options to fix
Option 1: Fix in DataFusion upstream (best long-term)
DataFusion's row group range pruning could be improved to assign each row group to exactly one split, e.g., by checking if the row group's byte range overlaps the split range rather than just checking the start
offset. This would be a contribution to the https://github.com/apache/datafusion project.
Option 2: Override row group selection in Comet's ParquetSource
Comet already creates a custom ParquetSource. You could implement a custom ParquetAccessPlan or row group filter that uses Spark's exact split boundaries to decide ownership. DataFusion's ParquetSource supports
with_row_group_filter() — you could provide a filter that says "only read row groups whose midpoint (or start of data) falls in my range," matching Spark's assignment logic.
Option 3: Pre-split at the Spark level to align with row groups
In CometNativeScan serialization, before sending ranges to native, adjust the ranges to align with Parquet row group boundaries. This would require reading Parquet metadata on the JVM side (which Spark already does
for count() — explaining why count() works correctly).
Option 4: Post-filter on native side
After DataFusion reads row groups, add deduplication logic so that when a row group spans two splits, only one split processes it. This is fragile but doesn't require upstream changes.
Most practical path
Option 2 is probably the most practical near-term fix. DataFusion's ParquetSource has hooks for customizing row group selection. You'd implement Spark's exact row group assignment logic: a row group belongs to a
split if its offset falls within [split.start, split.start + split.length). This way, each task reads exactly the row groups Spark intended it to read, and no task ends up idle.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] `native_datafusion` doesn't use all available parallelism for scan [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on issue #3817:
URL: https://github.com/apache/datafusion-comet/issues/3817#issuecomment-4193279630
Spark's split logic guarantees each row is counted exactly once. The `prune_by_range check` is a performance optimization, not a correctness gate. If a row group's start
offset falls within the range, that executor reads it; otherwise it skips it. Since RG1 starts at ~122MB (< 128MB split point), chunk 1 always claims it, and chunk 2 never does.
The problem is performance waste:
```
┌──────────┬─────────────────────────────┬────────────────────────┐
│ │ Chunk 1 (0..128MB) │ Chunk 2 (128MB..190MB) │
├──────────┼─────────────────────────────┼────────────────────────┤
│ Intended │ Read RG0 (~273K rows) │ Read RG1 (~152K rows) │
├──────────┼─────────────────────────────┼────────────────────────┤
│ Actual │ Reads RG0 + RG1 (425K rows) │ Reads nothing (0 rows) │
└──────────┴─────────────────────────────┴────────────────────────┘
```
- Chunk 1 executors do ~1.5x the intended work (read both row groups)
- Chunk 2 executor wastes ~800ms loading metadata only to return 0 rows
- Across 1800 partitions, this means hundreds of tasks doing zero useful work
Root cause: DataFusion's `prune_by_range` decides row group ownership by checking if col0_dict_page_offset falls within [range.start, range.end). When both RGs start before the 128MB split point, chunk 1
gets both and chunk 2 gets none.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] `native_datafusion` doesn't use all available parallelism for scan [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on issue #3817:
URL: https://github.com/apache/datafusion-comet/issues/3817#issuecomment-4158113473
with https://github.com/apache/datafusion-comet/pull/3842 I can see 1200 tasks not doing the actual scan, so the problem is in the task assignment/scheduler
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] `native_datafusion` doesn't use all available parallelism for scan [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on issue #3817:
URL: https://github.com/apache/datafusion-comet/issues/3817#issuecomment-4203253669
The workaround is to play `spark.sql.files.maxPartitionBytes` to 64M which is half of default value.
Fewer this setting the fewer the difference between Spark/Comet number of tasks that reads data. However the smaller setting it generates more idle scan tasks for both Spark/Comet. I'm not currently sure if PR is needed to solve this
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] `native_datafusion` doesn't use all available parallelism for scan [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #3817:
URL: https://github.com/apache/datafusion-comet/issues/3817#issuecomment-4246195022
@comphead we should document all of this in the tuning guide
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] `native_datafusion` doesn't use all available parallelism for scan [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on issue #3817:
URL: https://github.com/apache/datafusion-comet/issues/3817#issuecomment-4145581815
Local HDFS cluster tests are splitting files correctly between tasks
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org