@@ -85,6 +85,14 @@ remove the inserted local node when restoring Spark execution. Restoring its `or
 apply the global TopK twice. A single input partition uses a final native limit and projection;
 multiple input partitions still require a final TopK after the shuffle.
 
+With `spark.comet.exec.topK.dynamicFilter.enabled`, only the local sort sets the protobuf
+`Sort.dynamic_filter_enabled` flag. The native planner wraps an eligible sort in
+`TopKReaderFilterExec`. Its permanent plan contains an unexecuted sort template; each execution
+creates a fresh sort and live predicate, then attaches that predicate to the Parquet reader when
+eligible. The stream owns the heap and predicate until completion, error, or cancellation. The
+final TopK after an exchange does not share this state. Reader work stays on scan metrics while
+attachment counters belong to the local TopK.
+
 ### Choosing the Right Operator Type
 
 When adding a new operator, choose based on these criteria:
 
@@ -27,13 +27,15 @@
 - Spark 3.5.8 (audited 2026-05-27): baseline. `Coalesce(children) extends Expression with ComplexTypeMergingExpression`; returns the first non-null child, evaluated left-to-right with short-circuit. Result type is the merged child type. Comet routes via `CometCoalesce`, which serialises as nested `CaseWhen(IsNotNull(c1) -> c1, IsNotNull(c2) -> c2, ..., else cN)` so the native engine preserves the short-circuit semantics.
 - Spark 4.0.1 (audited 2026-05-27): byte-for-byte identical to 3.5.8.
 - Spark 4.1.1 (audited 2026-05-27): byte-for-byte identical to 3.5.8.
+- Performance (tuned 2026-09-28, PR [#6350](https://github.com/apache/datafusion-comet/pull/6350)): runs through the `CASE WHEN` evaluation described under `when`. 12-13x faster for two columns. Benchmark: `benches/conditional.rs`.
 
 ## if
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
 - Spark 3.5.8 (audited 2026-05-27): baseline. `If(predicate, trueValue, falseValue) extends ComplexTypeMergingExpression`; standard ternary semantics with short-circuit, predicate must be `BooleanType`. Comet routes via `CometIf` to the native `IfExpr` proto.
 - Spark 4.0.1 (audited 2026-05-27): adds an `override def withNewAlwaysEvaluatedInputs(...)` hook for the new optimizer phase; semantics unchanged. Error messages reformatted (`paramIndex` -> `ordinalNumber`).
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
+- Performance (tuned 2026-09-28, PR [#6350](https://github.com/apache/datafusion-comet/pull/6350)): runs through the `CASE WHEN` evaluation described under `when`, with an `IF` or `CASE WHEN` in the ELSE flattened into one merge. 1.5x faster with literal results, 6.6x with column results, 3.3-4.7x with string columns, 3.1-5.9x nested. Benchmark: `benches/conditional.rs`.
 
 ## ifnull
 
@@ -69,5 +71,6 @@
 - Spark 3.5.8 (audited 2026-05-27): the `CASE WHEN ... THEN ...` SQL form lowers to `CaseWhen(branches: Seq[(Expression, Expression)], elseValue: Option[Expression])`. Spark evaluates left-to-right with short-circuit; result type is the merged branch type. Comet routes via `CometCaseWhen` to the native `CaseWhen` proto.
 - Spark 4.0.1 (audited 2026-05-27): adds the `withNewAlwaysEvaluatedInputs` optimizer hook; semantics unchanged.
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
+- Performance (tuned 2026-09-28, PR [#6350](https://github.com/apache/datafusion-comet/pull/6350)): when no branch value and no WHEN after the first can fail, `CaseWhenExpr` evaluates them over the whole batch and takes each row's value with typed merges for primitive, boolean, string and binary results, instead of filtering the batch for each branch and merging the partial results with `MutableArrayData`. A branch is no longer wrapped in a cast to the type it already has, which hid literals and columns from DataFusion's fast paths. 2.8-4x faster with literal results and 2.2-7.8x with column results. When a branch can fail it is about 2% slower, because DataFusion's `CaseExpr` derives the result type of a bare branch for every batch. Benchmark: `benches/conditional.rs`.
 
 [Spark Expression Support]: ../../user-guide/latest/expressions.md
@@ -107,6 +107,30 @@ by the reader. An attached filter does not guarantee that any row groups are pru
 probe scan's `bytes_scanned` and `row_groups_pruned_statistics` with filtering disabled to assess
 reader savings. Existing join, scan, and intervening filter metrics retain their own meanings.
 
+### Local TopK
+
+With `spark.comet.exec.topK.dynamicFilter.enabled=true`, eligible fused local TopK operators report
+these counters. See [TopK Reader Pruning](tuning/operators.md#topk-reader-pruning) for the required
+fusion option and reader restrictions.
+
+| Metric                                 | Description                                                                  |
+| -------------------------------------- | ---------------------------------------------------------------------------- |
+| `dynamic_filter_topk_filters_attached` | Executions that attach their live TopK threshold to a native Parquet reader. |
+| `dynamic_filter_topk_filters_skipped`  | Eligible TopK executions whose input cannot accept reader attachment.        |
+
+These counters belong to the local TopK. An attached predicate may prune nothing, and a file's
+schema adaptation can disable it to preserve conversion errors. Reader work remains on the scan:
+`output_rows` counts rows emitted by the scan, `bytes_scanned` measures requested data/Bloom-filter
+ranges (excluding footer and page-index reads), and page and decoder filtering use the existing
+`page_index_rows_pruned` and `pushdown_rows_pruned` counters.
+
+Row groups skipped by runtime pruning within a file increment `row_groups_pruned_dynamic_filter`.
+When later files open, the reader can prune using the TopK threshold already available; those groups
+increment `row_groups_pruned_statistics`. With one row group per file, the dynamic counter can remain
+zero even when TopK skips most later groups. The statistics counter also includes pruning by other
+predicates, so compare both counters and `bytes_scanned` with filtering disabled to assess TopK savings.
+This feature adds no separate decoded-batch filter or evaluated/pruned row counters to the TopK.
+
 ### Exchange
 
 Comet adds some additional metrics:
@@ -185,6 +209,7 @@ execution metrics. Counters accumulate per scan operator; they do not instrument
 | `scan_io_object_store_response_bytes_read` | Response bytes actually consumed at that API, including bytes fetched between coalesced ranges. Not HTTP wire bytes.                                                                                                         |
 | `scan_io_metadata_cache_hits`              | Successful, cache-eligible metadata opens requiring no storage reads.                                                                                                                                                        |
 | `scan_io_metadata_cache_misses`            | Successful, cache-eligible metadata opens requiring storage reads. Failed opens and encrypted opens, which bypass this shared cache, increment neither cache counter.                                                        |
+| `row_groups_pruned_dynamic_filter`         | Row groups skipped by a runtime predicate whose value changes during execution, such as a local TopK threshold.                                                                                                              |
 
 Reader-level and object-store bytes are two views of the same reads; do not add them together.
 Likewise, footer bytes are a subset of metadata bytes, not a third reader-level category. A warm
 
@@ -111,12 +111,44 @@ Each scan partition keeps enough candidates for both `LIMIT` and `OFFSET`. With
 Comet shuffles those candidates and performs the final TopK. With one partition, it reuses the local
 ordering without building a second heap. The final stage applies the offset and output projection.
 
-Fusion reduces the work of passing scan batches between native execution blocks. It still reads all
-input rows and does not enable TopK reader pruning. It can also reduce overlap between scan decoding
+Fusion reduces the work of passing scan batches between native execution blocks. Fusion alone still
+reads all input rows. [TopK reader pruning](#topk-reader-pruning) is enabled separately. Fusion can
+also reduce overlap between scan decoding
 and TopK processing, so some workloads may run slower. Compare enabled and disabled runs with your
 data layout, payload width, limit, and partition count before enabling it. The
 `CometTopKBenchmark` microbenchmark covers these cases with ascending, descending, and random layouts.
 
+### TopK Reader Pruning
+
+Set both `spark.comet.exec.topK.fusion.enabled=true` and
+`spark.comet.exec.topK.dynamicFilter.enabled=true` to pass the local TopK's improving threshold to
+its Parquet reader. Both options are experimental and disabled by default. Eligibility is the same
+single signed integer key described above. Each task creates a fresh threshold; it is not shared
+across Spark partitions or exchanges, or retained for later executions.
+
+Once the heap contains enough candidates for `LIMIT + OFFSET`, the reader can skip later row groups
+whose statistics prove that no row can improve those candidates. Existing Parquet page-index and
+decoder-filter options can also use the predicate. This option adds no separate filter over decoded
+scan batches. TopK continues to select the final candidates.
+
+Reader attachment is conservative. A scan with a fetch limit, supplied file statistics, or a static
+predicate other than direct column `IS NOT NULL` checks keeps the existing execution path. For each
+file, schema adaptation disables pruning if it could hide a conversion error in a projected or
+filtered column. Missing null counts remain unknown, which can prevent pruning even when min/max
+statistics are present. These cases can still execute a fused TopK.
+
+Reader pruning is most useful when small K values and the file order establish a strong threshold
+early. Descending or random layouts for an ascending query can prune few or no groups, while still
+paying the cost of attaching and checking the predicate. Wider rows can increase the benefit when
+groups are skipped. Compare `pruning` with `fused` in `CometTopKBenchmark` to measure the reader effect,
+and compare both with `unfused` to include the cost of fusion. Check the scan's emitted rows,
+`bytes_scanned`, `row_groups_pruned_dynamic_filter`, and `row_groups_pruned_statistics` alongside elapsed
+time; attachment alone does not demonstrate a saving. Pruning when later files open uses the TopK
+threshold already available and increments `row_groups_pruned_statistics`. With one row group per
+file, the dynamic counter can stay zero despite substantial TopK pruning. The statistics counter also
+includes other predicates, so compare with filtering disabled to assess TopK savings.
+See [TopK metrics](../metrics.md#local-topk).
+
 ## Optimizing Sorting on Floating-Point Values
 
 Comet normalizes NaN payloads and signed zeros in scalar `FLOAT` and `DOUBLE` ordering keys, so `ORDER BY`, window
 
@@ -19,8 +19,10 @@
 
 mod join;
 mod parquet_reader;
+mod topk;
 
 pub(crate) use join::DynamicFilterJoinExec;
+pub(crate) use topk::TopKReaderFilterExec;
 
 use std::fmt::Formatter;
 use std::sync::Arc;
 
@@ -41,7 +41,7 @@ use schema_adapter::RuntimeFilterSchemaAdapterFactory;
 /// or changing the predicate. Every accepted leaf is deterministic, infallible,
 /// and only discards rows, so reader pruning cannot suppress expression errors
 /// or alter stateful evaluation. All other expressions remain a boundary.
-fn is_direct_column_null_checks(predicate: &Arc<dyn PhysicalExpr>) -> bool {
+pub(super) fn is_direct_column_null_checks(predicate: &Arc<dyn PhysicalExpr>) -> bool {
     if let Some(binary) = predicate.downcast_ref::<BinaryExpr>() {
         return binary.op() == &Operator::And
             && is_direct_column_null_checks(binary.left())