Repository navigation
[pull] main from apache:main - #262
Merged
Merged
Conversation
…ple probe partitions (#22791) ## Problem `NestedLoopJoinExec` can return incorrect and non deterministic results for LEFT, RIGHT and FULL joins when the probe side has more than one partition (the common case for any join whose condition is not a pure equijoin). Some unmatched left rows are emitted twice: once correctly and once again as an extra NULL padded row. ## Proof Correct answer is 5 (every t1 row is unmatched). Verify with `target_partitions = 1`. With multiple partitions an unpatched build intermittently returns 6, 7 or 8. Because it is a scheduling race, run it a few times to observe the divergence. ```sql set datafusion.execution.target_partitions = 4; set datafusion.execution.batch_size = 2; create table t1(v bigint) as values (4),(72),(41),(98),(91); create table t2(s varchar, w bigint) as values ('aaaaaaaaaaaaaaaa', 1),('bbbbbbbbbbbbbbbb', 2),('cccccccccccccccc', 3); select count(*) as left_join_rows from t1 left join t2 on (t1.v < t2.w and t2.s <= 'A'); ``` ## Solution The probe streams share a `probe_threads_counter`; the stream that drives it to 0 is the one that emits unmatched left rows, after all partitions finish probing. In `handle_emit_left_unmatched`, after `process_left_unmatched` returns `Ok(false)`, `maybe_flush_ready_batch()` can return early with a ready batch before the state advances to `Done`. The stream stays in `EmitLeftUnmatched`, so the next poll re enters `process_left_unmatched` with `left_emit_idx == 0` and decrements the counter a second time. The counter then reaches 0 before every partition has finished probing, so a partition emits unmatched left rows early. The fix decrements the counter at most once per probe stream using a new `probe_completed_reported` flag (reset per chunk in the memory limited path). All 42 existing `nested_loop_join` tests pass. This is separate from #22641, which covers the memory limited spill fallback path; this bug reproduces with unbounded memory and with spilling disabled.
…istinct values with many groups (#22768) ## Which issue does this PR close? - Closes #22767 ## Rationale for this change `approx_distinct` is very slow with `GROUP BY` on high-cardinality keys. On a dataset (~3.9M rows, ~512K groups), one file from the dataset describe in #22767 ```sql SELECT client_ip, approx_distinct(trace_id) AS cnt FROM '*.parquet' GROUP BY client_ip ORDER BY cnt DESC LIMIT 10; ``` - DataFusion: **~32.6s** - DuckDB (`approx_count_distinct`): **~0.1s** The reason is that `approx_distinct` only implemented `Accumulator`, not `GroupsAccumulator`. So grouped queries fell back to `GroupsAccumulatorAdapter`, which allocates a full 16 KiB HyperLogLog per group (~8 GB for 512K groups) and re-slices the input per group on every batch — even though most groups only see a few distinct values. ## What changes are included in this PR? - Add a dedicated `GroupsAccumulator` for `approx_distinct` that processes each batch in a single pass (no per-group slicing or dynamic dispatch). - Use an adaptive per-group sketch: keep a small list of hashes (sparse) and only switch to a dense 16 KiB HyperLogLog after 256 distinct values. This cuts memory and keeps the partial state small. The dense format stays compatible with the existing scalar accumulator. - Add `count_from_hashes` so small groups are estimated directly from their stored hashes, avoiding a 16 KiB alloc + scan per group at output time. - Hashing matches the existing per-type scalar accumulators, so results are unchanged. Boolean / small-int / `Null` keep using the old path. Result on the query above: **~32.6s → ~0.12s** (~270x, on par with DuckDB), with identical output. ## Are these changes tested? Yes. - New unit tests for the per-group sketch (sparse/dense, promotion, serialize/merge round-trip, merging groups, empty groups), checked against a dense-fold reference. - New `aggregate.slt` cases: grouped `approx_distinct` over `Utf8`, `Utf8View`, and `Int32` (small groups are exact), null-only groups (= 0), and a sparse→dense case (2000 distinct/group, within HyperLogLog error). - Existing `aggregate.slt` and `aggregate_skip_partial.slt` still pass; clippy and fmt are clean. ## Are there any user-facing changes? No API or result changes — only a large speedup for `approx_distinct` with `GROUP BY` on high-cardinality keys.
## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> - part of #22710 - 1/N of #22712 ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. --> See issues. This PR split out partial and final aggregate strem from `GroupsHashAggregateStream` To fully migrate hash aggregation, we have to - Port this optimization back #11627 - Support spilling I think they should be leave to follow up PRs Todo in this PR: - [x] Add a temporary configuration `enable_migration_aggregate` to turn off this path Since it should be a regression if the above features are not added, it also helps if to prevent potential regressions from the migration of other aggregate streams. ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR. --> Split out the streams from `GroupsHashAggregateStream` 1. Partial stage of hash aggregation 2. Final stage of hash aggregation ## Are these changes tested? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 3. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> ## Are there any user-facing changes? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. --> <!-- If there are any breaking changes to public APIs, please add the `api change` label. -->
## Which issue does this PR close? - N/A ## Rationale for this change Add new spark function: https://spark.apache.org/docs/latest/api/sql/index.html#quote ## What changes are included in this PR? - Implementation - SLT tests ## Are these changes tested? Yes, tests added as part of this PR. ## Are there any user-facing changes? No, these are new function. --------- Co-authored-by: Kazantsev Maksim <mn.kazantsev@gmail.com>
## Which issue does this PR close? - Closes #22249. - Closes #17853. ## Rationale for this change Optimize projections for the static and recursive terms of recursive CTEs just like regular subqueries. The previous implementation optimized projections based on the outer columns, which could cause bugs. This new implementation still ensures that #16684 remains fixed. ## What changes are included in this PR? - Updated `optimize_projections` to be applied to the static and recursive terms of a recursive query. - Updated and added tests. ## Are these changes tested? Yes. ## Are there any user-facing changes? Maybe if the user relied on projections being pushed down from the outer query to the recursive CTE, but this can be fixed by removing those unnecessary projections directly in the CTE. --------- Co-authored-by: Bruce Ritchie <bruce.ritchie@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
See Commits and Changes for more details.
Created by
pull[bot] (v2.0.0-alpha.4)
Can you help keep this open source service alive? 💖 Please sponsor : )