GH-3772: Bulk skip in RunLengthBitPackingHybridDecoder / DictionaryValuesReader

[abstractdog]

Rationale for this change

ValuesReader.skip(int n) defaults to a naive loop of skip() calls. For dictionary-encoded columns each skip() bottoms out in RunLengthBitPackingHybridDecoder.readInt() — a mode switch, array indexing, and a value the caller immediately discards. Any filter-then-skip path (column-index row ranges, hash-join probe filtering, runtime filters) pays this cost per skipped row.

RleSkipBenchmark, JMH throughput, JDK 17, 100 k values/op:

pattern	bitWidth	readInt() loop	skipInts	speedup
rle	8	1.38 B/s	104.9 B/s	~76×
packed	8	0.89 B/s	4.12 B/s	~4.6×
mixed	8	0.96 B/s	6.99 B/s	~7.2×

What changes are included in this PR?

• New RunLengthBitPackingHybridDecoder.skipInts(int) — re-uses readNext() per run, then advances currentCount by min(n, currentCount) instead of walking every value through readInt().
• skip(int) overrides on DictionaryValuesReader and RunLengthBitPackingHybridValuesReader delegating to decoder.skipInts(n).
• parquet-benchmarks / RleSkipBenchmark JMH benchmark.

readNext() and the ValuesReader.skip(int) default are unchanged.

Are these changes tested?

TestRunLengthBitPackingHybridDecoderSkip covers RLE-only, PACKED-only, mixed, zero-skip, full-skip, mid-run partial skip, bitWidth = 0, and a randomised 4 K-value read/skip interleaving cross-checked against a readInt() reference. All parquet-column tests pass.

Are there any user-facing changes?

No. Additive and binary-compatible: no signatures change; skipInts(n) is semantically identical to N discarded readInt()s. Existing callers of ValuesReader.skip(int) pick up the fast path with no code change.

[abstractdog]

can I ask for a review of this?

made Hive-side benchmark with this change, wonderful results for dictionary columns, see details in apache/hive#6758 (comment)

[Fokko]

Nice concise optimization :)

[divjotarora]

Nice optimization @abstractdog! One small comment to maintain today's exception behavior in the empty dictionary case.

[divjotarora]

DictionaryValuesReader#initFromPage has this code:

public void initFromPage(int valueCount, ByteBufferInputStream stream) throws IOException {
    this.in = stream.remainingStream();
    if (in.available() > 0) {
      LOG.debug("init from page at offset {} for length {}", stream.position(), stream.available());
      int bitWidth = BytesUtils.readIntLittleEndianOnOneByte(in);
      LOG.debug("bit width {}", bitWidth);
      decoder = new RunLengthBitPackingHybridDecoder(bitWidth, in);
    } else {
      decoder = new RunLengthBitPackingHybridDecoder(1, in) {
        @Override
        public int readInt() throws IOException {
          throw new IOException("Attempt to read from empty page");
        }
      };
    }
  }

Specifically in the else, we create a special RunLengthBitPackingHybridDecoder for empty dictionary pages. Should we override that decoder's skipInts to preserve the same IOException behavior when skipping? Otherwise, we bypass readInt and call readNext in the skip path, which throws IllegalArgumentException.

[abstractdog]

makes sense @divjotarora , good catch, thanks! addressed it

[divjotarora]

Nice change @abstractdog!

[divjotarora]

I prefer not to have comments like "used to run" in code as it's codifying old behavior that's been replaced. Is this actually necessary here? Seems fine to just say "Skipping zero values does nothing and stays silent" without any mention of the old loop.

This is personal preference, so feel free to ignore if you disagree / defer to the maintainers of the repo.

[abstractdog]

agree, tend to take care of the same, missed this time:
80ecda7