Parquet Java ALP Implementation

[vinooganesh]

cc @julienledem @alamb @emkornfield @prtkgaur

Rationale for this change

Reworks the ALP encoding implementation to address emkornfield's architectural feedback on PR #3390. The original buffered all values in memory and decoded eagerly. This makes the writer incremental (encode per-vector as values arrive) and the reader lazy (decode on demand), matching how other Parquet encodings work.

Builds on Julien Le Dem's original implementation (#3390). File structure, integration points, core math, and interop test infrastructure all come from his work. The rework focused on the internal writer/reader plumbing.

What changes are included in this PR?

Architecture (addressing review feedback):

• Incremental writer. Values buffer in a fixed-size vector, each full vector encodes and flushes immediately.
• Lazy reader. Vectors decode on first access via offset array, skip() is O(1).
• Interleaved page layout so each vector is self-contained.
• Extracted AlpValuesReader abstract base class for shared logic.
• Preset caching. Full parameter search for first 8 vectors, top 5 combos cached for the rest.

Spec compliance:

• Fixed packed data size formula to ceil(n * bitWidth / 8)
• Fixed unsigned delta comparison in float writer
• Explicit little-endian byte reads instead of relying on ByteBuffer order
• Using parquet-encoding's BytePacker instead of custom bit-packing
• Capped max vector size at 32768 to prevent uint16 overflow in num_exceptions

Configurable vector size (new):

• withAlpVectorSize(int) and withAlpVectorSize(String columnPath, int) on ParquetProperties.Builder and the Hadoop ParquetWriter.Builder
• Threaded through DefaultV1ValuesWriterFactory and DefaultV2ValuesWriterFactory so per-column overrides work
• Defaults to 1024; validated against AlpConstants min/max bounds eagerly at builder time
• Enables generating fixture data at multiple vector sizes for cross-language compatibility testing

Reader null tolerance (bug fix):

• AlpValuesReader was asserting num_elements == page.valuesCount, which fails on optional columns with nulls (num_elements is the encoded non-null count, valuesCount is the page row count including nulls). Relaxed to num_elements > valuesCount. Surfaced by the corner-case fixture below.

Test fixtures (parquet-testing #100 prep):

• generateAlpFixturesAtMultipleVectorSizes reads the source parquets from parquet-testing PR PARQUET-134 patch - Support file write mode #100 and re-encodes them as Java ALP across 3 axes — page version (V1/V2), vector size (1024/4096), and dataset/type — producing 16 files, each verified bit-exact against its source.
• generateAndVerifyCornerCaseFixture writes a small (~60 KB) synthetic file with 14 columns each engineered to hit a specific corner case from parquet-testing Parquet 167 - Snappy Compression codec - Optimize freeing of DirectByteBuffers #105: no/one/all exceptions, NaN/Inf/-0.0, constant (bit_width=0), differing exponents per vector, optional with nulls. Both f32 and f64.
• writeCornerCaseCsvTruth emits a sidecar _expect.csv from the construction recipe (not from reading the parquet back), so the corner-case file is independently verifiable by any future tooling.
• readAllFixtureFilesIndependently opens every generated fixture and asserts ALP appears in column encodings + every row decodes through the standard reader path.

Integration:

• Wired ALP into both DefaultV1ValuesWriterFactory and DefaultV2ValuesWriterFactory.

Are these changes tested?

Yes. 125 tests across 5 test classes in parquet-column, plus 12 tests in TestInterOpReadAlp in parquet-hadoop — all passing. Full parquet-column suite also passes.

Key tests construct ALP page bytes directly according to the spec and feed them to the reader without going through the writer. This verifies the reader works independently and catches any bugs where writer and reader agree with each other but disagree with the spec. Also covers NaN bit pattern preservation, negative zero roundtrip, extreme values, every partial vector remainder mod 8, skip across vector boundaries, and preset caching under distribution change.

Cross-language verification

Arrow C++ ALP decoder (apache/arrow#48345) reads every Java-written fixture bit-exact against the canonical _expect.csv truth tables from parquet-testing PR #100. Local verification covers all 17 fixtures across the full {V1, V2} × {vs1024, vs4096} matrix: 1,587,306 values, 0 mismatches. Six representative fixtures submitted as a stacked PR (prtkgaur/parquet-testing#1) toward apache/parquet-testing#100 account for 342K of those.

Are there any user-facing changes?

• Users can enable ALP encoding for FLOAT and DOUBLE columns via ParquetProperties.withAlpEncoding(), globally or per-column.
• Users can configure the ALP vector size via withAlpVectorSize(int) (new in this PR), also globally or per-column. Default is 1024.

Note: Likely me missing something — but ALP is not yet in the parquet-format Thrift spec (apache/parquet-format#533), so writing ALP files through the full Hadoop pipeline will fail at metadata serialization until parquet.thrift is updated (parquet-format PR #548).

[prtkgaur]

The code organization looks good to me and the code follows the spec. I looked for areas of any extra buffer allocations which might impact performance and I think it is optimally written.

I think we should add a few benchmarks and publish numbers from them.

Thanks for working on this Vinoo!

[prtkgaur]

Wanted to make sure we have the following testing.

For the cross compatibility testing are we making sure that we write both V1 and V2 pages and the implementation in other language is able to read it.

[RussellSpitzer]

Should we check that the bitwidth is also within reasonable bounds? bitWidth within [0,32] or [0,64] respectively?

[RussellSpitzer]

I don't see a test for this or the float factory producing the ALPValueWriter classes. Probably worth adding them just like the ByteStreamSplit tests in DefaultValuesWriterFactoryTest

[RussellSpitzer]

I'm not sure if this belongs in DefaultV1ValuesWriter factory, we'll see what @rdblue proposes with the new versioning. But I"m guessing this will land in a V3? I would stick it only in V2 for now since that's already a bit mixed up

[RussellSpitzer]

May want to add a case for numElements < Value count to cover the case where some values where null.

[RussellSpitzer]

nit: The Naming is a bit odd for me here

As the javadoc notes, this only checks for blanket exceptions but the method below
is the one that would tell us if the value was actually an exception.

Just feels a little worrysome to me that you could see isFloatException(value) : false
but isFloatException(value, x, y) : true

I might name this something different to make it clear that it's a subset of checks.

maybe isIntrinsicFloatException(floatValue)
and isFloatException(value, exponent, factor)

[RussellSpitzer]

(this also would mean you could probably drop the javadocs explaining the difference)

[RussellSpitzer]

nit: Is this java doc really valuable here? I'm not sure it does more than describe the implementation.

[RussellSpitzer]

Same note on the polymorphism here

[vinooganesh]

Thanks @RussellSpitzer and @emkornfield! I'll do a first pass on the feedback ASAP

[RussellSpitzer]

Is this just here to make the spec explicit?

[RussellSpitzer]

Not sure we need the note about the C++ ALP cost model. Feels like that information could get stale and doesn't really help the reader.

[RussellSpitzer]

This I think would be nice to be metioned in the java doc. We have an obvious ordering (lowest estimated size wins) but then the decision for best e and f is I assume to just match what was done in the C++ impl?

[RussellSpitzer]

nit: I like doing constants for this kind of indexing just so we aren't constnatly remembering the constants here so

final int E = 0
final int F = 1

But that's just me

[RussellSpitzer]

Same sign question as the above function

[RussellSpitzer]

I'm a little worried about the code duplication here. I think we are directly
duplicating some of the logic here.

What if instead we did something like

scoreFloat(buffer, offset, length, e, f)

pickBestFloat(buffer, offset, length, pairs[])

findBestFloat(buffer, offset, length):
    return pickBestFloat(buffer, offset, length, ALL_VALID_PAIRS)

findBestFloat(buffer, offset, length, presets):
    return pickBestFloat(buffer, offset, length, presets)

[RussellSpitzer]

I think this is only true for get(). We should have LE respecting getShort, getInt and getLong on the buffer since we already slice it as little littleEndian.

Basically if we make

    this.vectorsData = rawSlice.slice().order(ByteOrder.LITTLE_ENDIAN);
    //Then     
   (vectorsData.getShort(pos) & 0xFFFF) == getShortLE(vectorsData, pos)
   vectorsData.getInt(pos) == getIntLE(vectorsData, pos)
   vectorsData.getLong(pos) == getLongLE(vectorsData, pos)

If we still want buffer agnostic helpers we already have some helpers in org.apache.parquet.bytes.BytesUtils. That would probably be the right place
to add more.

  /**
   * reads an int in little endian at the given position
   *
   * @param in     a byte buffer
   * @param offset an offset into the byte buffer
   * @return the integer at position offset read using little endian byte order
   * @throws IOException if there is an exception reading from the byte buffer
   */
  public static int readIntLittleEndian(ByteBuffer in, int offset) throws IOException {

SO I think we would put additional ones there as well

TLDR; I don't think we need these helpers here since we have a littleEndian buffer already, but it f were going to add helpers we should do them in ByteUtils

[RussellSpitzer]

Reading through this and the subclasses I find it a little confusing that we have so many different "index" cursors used with different names
in different contexts. I think they are a little confusing because they are in essentially different coordinate systems.

I think it may be clearer if we give them some slightly different names

currentIndex -> pageValueIndex // Which encoded value in the page
vectorIdx / currentVectorIndex -> vectorNumber // Which vector chunk 
indexInVector -> vectorSlot // Position within that chunk

To clearly differentiate what each one represents

[RussellSpitzer]

I mentioned on AlpValuesReaders but this is where I think it would be much clearer if we didn't use "Idx" and instead used something like vectorNumber

[RussellSpitzer]

The 0XFFFF should be redundant here with the current impl of getShortLE (it already masks both get's with 0XFF), but if we switch to vectorsData.getShort(pos + 2) we would need it

[RussellSpitzer]

nit: pos + Long.Bytes ?

[RussellSpitzer]

BytesUtils has paddedByteCountFrombits which has this same math

[RussellSpitzer]

Looking back there is a BytesUtils for this as well getWidthFromMaxInt(maxDelta)

[RussellSpitzer]

I left a bunch of other comments on the double reader, they all would apply here as well s/double/float and s/long/int/ :)

[RussellSpitzer]

Do we want a bounds check for bitWidth <= 64?

[RussellSpitzer]

Very much my personal opinon , but it feels a little weird to me that the "cached" version comes first and the "build cache" version comes second.

i'm wondering if it would be simpler like

  if (cachedPresets == null) {
    // Sampling amd potential cache building first
    params = AlpEncoderDecoder.findBestFloatParams(vectorBuffer, 0, vectorLen);
    if (vectorsProcessed % rowgroupSampleJump == 0
        && rowgroupSamples.size() < SAMPLER_SAMPLE_VECTORS_PER_ROWGROUP) {
      rowgroupSamples.add(Arrays.copyOf(vectorBuffer, vectorLen));
    }
    if (rowgroupSamples.size() >= SAMPLER_SAMPLE_VECTORS_PER_ROWGROUP) {
      buildPresetCache();
    }
  } else {
    // Cached case second
    params = AlpEncoderDecoder.findBestFloatParamsWithPresets(
        vectorBuffer, 0, vectorLen, cachedPresets);
  }
  vectorsProcessed++;

This would tie the cache building to the sampling phase rather than being a branch either
phase could hit and would match the normal flow of code imho.

[RussellSpitzer]

I was walking through this and I think we are being a little more clever then this here right?

Maybe the comment could read more like

Exception indices still participate in the packed integer array (one slot per vector position), 
but the real float is stored in the exception block. We fill those slots with a dummy encoded int: 
  the first non-exception value's encoding if any exists, 
  otherwise 0 when the whole vector is exceptions.

Picking an encoding that's already in the vector avoids introducing a spurious minimum (e.g. 0) 
that would distort FOR/bit width. When every value is an exception, all slots are 0, minValue is 
0, and deltas are also 0.  The reader patches exception slots from the exception payload, so the 
placeholder only affects FOR/bit width, not the decoded values.

[RussellSpitzer]

Suggested change

	int bitWidth = AlpEncoderDecoder.bitWidthForInt(maxDelta);
	int bitWidth = BytesUtils.getWidthFromMaxInt(maxDelta);

[RussellSpitzer]

BytesUtils.paddedByteCountFromBits

[RussellSpitzer]

style nit: probably could use some line breaks between the for blocks here

[RussellSpitzer]

So for this function it feels like we are doing some redundant work

Instead of copying samples and finding best presets again, we could just save our sample
presets as we are sampling and then just run the frequency count here.

// at sample time
sampleWinners.add(new int[] { params.exponent, params.factor });

// buildPresetCache — no findBest, no float[] copies
private void buildPresetCache(sampleWinners int[][] ) {
Map<Long, Integer> counts = new HashMap<>();
for (int[] ef : sampleWinners) {
  long key = ((long) ef[0] << Integer.SIZE) | ef[1];
  counts.merge(key, 1, Integer::sum);
}
// sort, take top 5 → cachedPresets
}

So rather than recomputing for each sample, we just save the (e, f) for each sample (instead of the sample) and then group and sort by frequency?

Also I wouldn't use sampleWinners as a variable name here, but it's getting late and my naming brain is running out of energy

[RussellSpitzer]

It's getting late and I'm off next week. I reviewed pretty much everything except the tests.

My biggest checks were

1. Looks like there is a bitwidth calculation bug for floats using an unsigned comparison to determine range
2. The buildPresetCache looks like it's doing too much work, I think we can get by just saving the (e,f) values rather than copying vectors and refinding their e,f paris
3. There seems to be some duplication of BytesUtils methods with ALP specific versions here that seem to be the same. Is there a reason for this? May be worth consolidating otherwise

[RussellSpitzer]

All my comments from above, but s/float/double/ and s/int/long/ :)

[RussellSpitzer]

I'm not sure if this belongs in DefaultV1ValuesWriter factory, we'll see what @rdblue proposes with the new versioning. But I"m guessing this will land in a V3? I would stick it only in V2 for now since that's already a bit mixed up

[RussellSpitzer]

Maybe we should be bundling these into ALPConfig rather than having two ColumnProperties

[RussellSpitzer]

The bitwidth bug I noted for floats could have a test here

@test
public void findBestFloatParamsMixedSignUsesSignedDeltaRange() {
float[] values = {-1.0f, 1.0f}; // e=0,f=0 → encoded {-1, 1}
AlpEncoderDecoder.EncodingParams params =
AlpEncoderDecoder.findBestFloatParams(values, 0, values.length);
assertEquals(0, params.exponent);
assertEquals(0, params.factor);
assertEquals(0, params.numExceptions);
}

[RussellSpitzer]

Not sure what this means? Is this because we are renaming the BytesUtil methods? I'm still not sure why we are reimplementing those here