[branch-4.1] Backport Iceberg, Paimon, and file scanner changes

be/benchmark/benchmark_main.cpp

@@ -45,6 +45,7 @@
 #include "core/data_type/data_type.h"
 #include "core/data_type/data_type_string.h"
 #include "parquet/benchmark_parquet_decoder.hpp"
+#include "parquet/benchmark_parquet_kernels.hpp"
 #include "parquet/benchmark_parquet_reader.hpp"
 #include "runtime/exec_env.h"
 #include "runtime/memory/mem_tracker_limiter.h"

be/benchmark/parquet/AGENTS.md

@@ -6,9 +6,10 @@ benchmark system described in the design document.
 
 ## What exists today
 
-The benchmark binary registers two groups:
+The benchmark binary registers three groups:
 
 - `ParquetDecoder`: native page decoder benchmarks using in-memory encoded pages.
+- `ParquetKernel`: isolated SIMD-sensitive decode and predicate kernels.
 - `ParquetReader`: local-file benchmarks that call the format V2 Parquet reader directly.
 
 The relevant files are:
@@ -36,13 +37,16 @@ List all Parquet cases and verify the expected registration counts:
 
 ```shell
 be/output/lib/benchmark_test --benchmark_list_tests \
-  | grep -E '^Parquet(Decoder|Reader)/'
+  | grep -E '^Parquet(Decoder|Kernel|Reader)/'
 
 be/output/lib/benchmark_test --benchmark_list_tests \
-  | grep -c '^ParquetDecoder/'  # currently 152
+  | grep -c '^ParquetDecoder/'  # currently 228
 
 be/output/lib/benchmark_test --benchmark_list_tests \
-  | grep -c '^ParquetReader/'   # currently 137
+  | grep -c '^ParquetKernel/'   # currently 80
+
+be/output/lib/benchmark_test --benchmark_list_tests \
+  | grep -c '^ParquetReader/'   # currently 152
 ```
 
 When running the binary directly from `be/build_RELEASE/bin`, make sure the JVM and third-party
@@ -61,6 +65,12 @@ be/output/lib/benchmark_test \
   --benchmark_out=parquet-decoder-smoke.json \
   --benchmark_out_format=json
 
+be/output/lib/benchmark_test \
+  --benchmark_filter='^ParquetKernel/' \
+  --benchmark_min_time=0.001s \
+  --benchmark_out=parquet-kernel-smoke.json \
+  --benchmark_out_format=json
+
 be/output/lib/benchmark_test \
   --benchmark_filter='^ParquetReader/' \
   --benchmark_min_time=0.001s \
@@ -98,8 +108,8 @@ cache to manufacture a cold run.
 
 ## Current scenario matrix
 
-`ParquetDecoder` contains 19 encoding/type pairs. Each pair is run at 1%, 10%, 50%, and 100%
-selection with clustered and alternating selection ranges, for 152 registered cases.
+`ParquetDecoder` contains 19 encoding/type pairs. Each pair is run at 0%, 1%, 10%, 50%, 90%, and
+100% selection with clustered and alternating selection ranges, for 228 registered cases.
 
 | Encoding | Physical types |
 |---|---|
@@ -110,10 +120,16 @@ selection with clustered and alternating selection ranges, for 152 registered ca
 | DELTA_LENGTH_BYTE_ARRAY | BYTE_ARRAY |
 | DELTA_BYTE_ARRAY | BYTE_ARRAY |
 
+`ParquetKernel` contains 80 cases across five SIMD-sensitive stages: BYTE_STREAM_SPLIT,
+DELTA_PREFIX_SUM, DICTIONARY_GATHER, NULLABLE_EXPAND, and RAW_PREDICATE. It covers the applicable
+four- and eight-byte types, three dictionary working-set sizes, 0% through 90% null rates with both
+placement patterns, and 0% through 100% raw-predicate selectivities.
+
 `ParquetReader` deliberately uses a single-variable matrix rather than a Cartesian product. After
-deduplication it contains 137 cases covering:
+deduplication it contains 152 cases covering:
 
-- operations: open-to-first-block, full scan, predicate scan, limit 1, and limit 1000;
+- operations: open-to-first-block, full scan, predicate scan, complex residual scan, limit 1, and
+  limit 1000;
 - file encodings: PLAIN, dictionary, BYTE_STREAM_SPLIT, and DELTA_BINARY_PACKED;
 - null ratios: 0%, 1%, 10%, 50%, and 90%;
 - null shapes: clustered and alternating;
@@ -146,10 +162,12 @@ generator, random seed, or manifest involved.
   and intentionally produces many one-row physical ranges.
 
 Page generation, selection construction, decoder creation, dictionary setup, and `set_data` are
-outside the timed decode call. The sinks consume decoder callbacks and prevent compiler removal,
-but they do not build a Doris `Column`. Consequently these cases isolate decoder traversal and
-selection cost; they do not measure definition-level decoding, nullable reconstruction, type
-conversion, or full column materialization.
+outside the timed decode call. Before timing, every decoder case verifies the consumed value count
+and a checksum of all selected values against the deterministic source generator. The timed sinks
+then consume decoder callbacks and prevent compiler removal, but they do not build a Doris
+`Column`. Consequently these cases isolate decoder traversal and selection cost; they do not
+measure definition-level decoding, nullable reconstruction, type conversion, or full column
+materialization.
 
 ## How reader Parquet files are generated
 
@@ -167,6 +185,9 @@ Fixture contents and writer settings are:
 - every non-null value is `row % 100`;
 - the predicate is `value < selectivity_percent`, so the threshold maps directly to the intended
   non-null selectivity;
+- the complex residual scan evaluates a production expression tree whose first child is
+  `c0 < selectivity_percent` and whose always-true second child is `c2 = c3`, exposing whether
+  later-only columns are decoded eagerly;
 - alternating nulls use a 101-row period and `(row * 37) % 101`, avoiding direct correlation with
   the 100-value predicate period;
 - clustered nulls use contiguous null prefixes inside each 1,024-row cluster;
@@ -234,13 +255,14 @@ The current matrix is not comprehensive. Preserve this distinction in PR descrip
 1. Add the design's `matrix.yaml`, deterministic corpus generator, `manifest.json`, checksum, and
    standalone corpus verifier. The current runtime-generated files cannot be shared unchanged with
    V1, StarRocks, or DuckDB.
-2. Add correctness oracles. Benchmarks must validate consumed counts and representative checksums
-   outside the timed region before their performance samples are trusted.
+2. Add full reader correctness oracles. Decoder cases validate consumed counts and selected-value
+   checksums outside the timed region, and kernel cases compare representative output. Reader cases
+   still need value checksums in addition to their output-row counters.
 3. Add reader-level INT64, FLOAT, DOUBLE, BYTE_ARRAY/string, FIXED_LEN_BYTE_ARRAY, DATE,
    TIMESTAMP, and DECIMAL cases. Today only nullable INT32 reaches the complete reader path.
-4. Extend decoder coverage with 0% and 90% selection, definition levels/null reconstruction,
-   dictionary conversion, and real Doris `Column` materialization. The current decoder sink does
-   not cover those costs or report decoded bytes per second.
+4. Extend decoder coverage with definition levels/null reconstruction, dictionary conversion, and
+   real Doris `Column` materialization. The current decoder sink does not cover those costs or
+   report decoded bytes per second.
 5. Add the representative nullable sparse corpus requested by the design: 32 INT64 columns, 128
    row groups, and enough rows to exercise many pages. The current 16K-row/four-row-group fixture
    is a smoke-sized workload.
@@ -277,10 +299,10 @@ be simulated by silently changing the local reader benchmark.
 
 ## Current validation record
 
-At commit `16e05dd5c71`, a Release build completed and the matrix unit test passed 4/4. A 1 ms smoke
-run executed 152 decoder and 137 reader cases with zero benchmark errors. This is an execution
-record only. It is not a reviewed performance baseline because repetitions, host isolation,
-warmups, cache control, `perf` data, variance, and before/after comparison were not collected.
+The current expected registration counts are 228 decoder, 80 kernel, and 152 reader cases. A smoke
+run is an execution record only, not a reviewed performance baseline, because repetitions, host
+isolation, warmups, cache control, `perf` data, variance, and before/after comparison are not
+collected.
 
 ## Rules for extending the suite
 

be/benchmark/parquet/README.md

@@ -28,21 +28,41 @@ timed region. It covers PLAIN, dictionary, byte-stream-split, and DELTA encoding
 supported fixed-width and binary physical types. Sparse selections are provided as both one
 clustered range and many alternating ranges.
 
+The decoder selection axis includes 0%, 1%, 10%, 50%, 90%, and 100% so boundary and
+high-selectivity behavior are visible.
+
 ```shell
 be/output/lib/benchmark_test \
   --benchmark_filter='^ParquetDecoder/plain/int64/sel_10/alternating$' \
   --benchmark_min_time=0.1s
 ```
 
+## SIMD kernel cases
+
+`ParquetKernel` isolates the five SIMD-sensitive stages from reader setup and virtual consumer
+overhead: byte-stream-split transpose, delta prefix sum, numeric dictionary gather, nullable
+expansion, and raw predicate evaluation. It covers the applicable 4-byte and 8-byte integer and
+floating-point physical types, raw-predicate selectivities from 0% through 100%, and nullable
+rates from 0% through 90% with clustered and alternating placement. Dictionary gather uses 32-,
+4,096-, and 262,144-entry working sets to separate cache-resident and cache-miss-dominated
+behavior.
+
+```shell
+be/output/lib/benchmark_test \
+  --benchmark_filter='^ParquetKernel/(dictionary_gather|nullable_expand)/' \
+  --benchmark_min_time=0.1s
+```
+
 ## Local reader cases
 
-`ParquetReader` measures local open-to-first-block, full scan, predicate scan, and LIMIT-shaped
-reads. The matrix covers:
+`ParquetReader` measures local open-to-first-block, full scan, predicate scan, complex residual
+scan, and LIMIT-shaped reads. The matrix covers:
 
 - PLAIN, dictionary, byte-stream-split, and DELTA binary-packed files;
 - NULL ratios of 0%, 1%, 10%, 50%, and 90%, with clustered and alternating placement;
 - predicate selectivities of 0%, 1%, 10%, 50%, 90%, and 100%;
 - predicate-only and predicate-plus-lazy-projected reads;
+- ordered complex residuals whose later columns are reachable only after an earlier residual;
 - schemas with 4, 32, 128, and 512 columns, with the predicate first or last.
 
 Fixtures are created lazily under the system temporary directory in
@@ -58,6 +78,16 @@ be/output/lib/benchmark_test \
   --benchmark_out_format=json
 ```
 
+The complex-residual case uses a production compound `AND` tree. Its first child,
+`c0 < selectivity_percent`, preserves the requested selectivity; its second child, `c2 = c3`,
+references two new columns and accepts every row that reaches it:
+
+```shell
+be/output/lib/benchmark_test \
+  --benchmark_filter='^ParquetReader/complex_residual_scan/plain/null_10/alternating/sel_10/' \
+  --benchmark_min_time=1s
+```
+
 Every result reports throughput plus `raw_rows`, `selected_rows`, `fixture_bytes`, `ns/raw_row`,
 and (when at least one row survives) `ns/selected_row`. Keep CPU frequency, build type, compiler,
 machine placement, and benchmark filters fixed when comparing two commits.