andygrove commented on code in PR #5110: URL: https://github.com/apache/datafusion-comet/pull/5110#discussion_r3676928167
########## docs/source/user-guide/latest/compatibility/operators.md: ########## @@ -19,6 +19,23 @@ under the License. # Operator Compatibility +## Sampling + +Comet runs `SampleExec` natively when sampling is performed without replacement, which covers +`DataFrame.sample`, SQL `TABLESAMPLE`, and `DataFrame.randomSplit`. The native implementation +reproduces Spark's per-row `XORShiftRandom` draw sequence, so for a given seed it selects the same +rows as Spark. + +Because the sampler consumes one random value per row within a partition, it selects the same rows +as Spark only when the rows reach it in the same order. Sampling directly above a scan, filter, or +projection meets that condition. Sampling above an operator where Comet may emit rows in a +different order than Spark, such as a join or an aggregate, still returns a valid sample of the +same expected size, but not necessarily the same rows Spark would have selected. + +Sampling with replacement (`df.sample(withReplacement = true, ...)`) falls back to Spark, because +it draws from a Poisson distribution that Comet does not implement natively Review Comment: Found it. The native sampling was in `native/core/src/execution/shuffle/range_partitioner.rs`: - Added by 3aa3dc707821ea7620c50658eb22af90b6b0edb7 (#1862, "feat: support RangePartitioning with native shuffle"). - Removed by 25d5924a72cfbe9f5effeda11f7f27eb09ec26eb (#2258, "fix: distributed RangePartitioning bounds calculation with native shuffle", closes #1906), because bounds were being computed per-executor and so were wrong in a distributed setting. That PR moved bounds calculation to the driver using Spark's own `RangePartitioner` and serialized the boundary rows to native. To recover the implementation plus its tests (474 lines): ``` git show 25d5924a7^:native/core/src/execution/shuffle/range_partitioner.rs ``` Two caveats on how much this helps the with-replacement path, though: 1. It is reservoir sampling (Algorithm L) for estimating range-partition bounds, i.e. sampling *without* replacement. That is a different problem from what `SampleExec` needs, which is a per-row Bernoulli or Poisson decision. 2. It seeded Rust's `rand::SmallRng` rather than porting Spark's RNG, so it produced a statistically valid sample but not Spark's exact draw sequence. That was fine for bound estimation, where agreement with Spark does not matter, but the `SampleExec` path needs exact agreement. So it is good prior art for having done sampling in native code, but it does not give us the `Well19937c` and `PoissonDistribution.sample()` ports that the with-replacement case still needs. I have noted the pointer on #5109 so the history is findable when we pick that up. -- 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]
