GH-3150: Keep column values intact with direct codecs

[1fanwang]

Rationale for this change

Reading the issue's Snappy file with a direct codec returns value_40001 in the key column instead of key_40001. The normal codec reads the same file correctly. After this fix, both readers return the expected values for all 51,000 rows.

Closes #3150.

What changes are included in this PR?

The page reader copies decompressed data and dictionaries into buffers it owns. A shared codec can then reuse its output buffer for another column without overwriting values still being read. Dictionary copies use the existing row-group cleanup. The path that decompresses directly into a reader-owned buffer is unchanged.

Are these changes tested?

Testing Done

Ran a standalone reader against the reporter's original file on macOS arm64 with JDK 17.0.5. The baseline is 2df8d02.

Reader	Before	After
Direct Snappy codec	Wrong key at row 40001; assertion fails	All 51,000 rows and both columns match
Normal codec	All values match	All values match

The baseline produced:

Exception in thread "main" java.lang.AssertionError: row=40001 key=value_40001 value=value_40001
	at ReadDirectCodec.main(ReadDirectCodec.java:24)

The fixed reader produced:

direct: verified 51000 rows and both column values

Built both standalone runtimes with Maven and Thrift 0.24.0. From this PR's checkout, these commands create the jars used below:

runtime=parquet-cli/target/parquet-cli-1.19.0-SNAPSHOT-runtime.jar
git worktree add --detach ../parquet-3150-baseline 2df8d02678dab4bb8b926a0d3221cc652984c7ab
(cd ../parquet-3150-baseline && ./mvnw -B -ntp -pl parquet-cli -am -Plocal -DskipTests package)
cp "../parquet-3150-baseline/$runtime" before-cli.jar
./mvnw -B -ntp -pl parquet-cli -am -Plocal '-Dtest=TestParquetReader,TestColumnChunkPageReadStore,TestDirectCodecFactory,ShowPagesCommandTest' -Dsurefire.failIfNoSpecifiedTests=false package
cp "$runtime" after-cli.jar

Download the fixture attached to the linked issue as reporter-test.parquet.zip, then extract it:

unzip -p reporter-test.parquet.zip test.parquet > test.parquet

Save the executed reader below as ReadDirectCodec.java, then run:

java -Xmx512m -XX:ActiveProcessorCount=2 -cp before-cli.jar ReadDirectCodec.java direct test.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -cp before-cli.jar ReadDirectCodec.java heap test.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -cp after-cli.jar ReadDirectCodec.java direct test.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -cp after-cli.jar ReadDirectCodec.java heap test.parquet

Executed reader

import org.apache.hadoop.fs.Path;
import org.apache.parquet.bytes.DirectByteBufferAllocator;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.hadoop.CodecFactory;
import org.apache.parquet.hadoop.ParquetReader;
import org.apache.parquet.hadoop.example.GroupReadSupport;

public class ReadDirectCodec {
  public static void main(String[] args) throws Exception {
    ParquetReader.Builder<Group> builder =
        ParquetReader.builder(new GroupReadSupport(), new Path(args[1]));
    if (args[0].equals("direct")) {
      builder.withCodecFactory(CodecFactory.createDirectCodecFactory(
          null, DirectByteBufferAllocator.getInstance(), 1024 * 1024));
    }
    int count = 0;
    try (ParquetReader<Group> reader = builder.build()) {
      Group record;
      while ((record = reader.read()) != null) {
        count++;
        String key = record.getString("key", 0);
        String value = record.getString("value", 0);
        if (!key.equals("key_" + count) || !value.equals("value_" + count)) {
          throw new AssertionError("row=" + count + " key=" + key + " value=" + value);
        }
      }
    }
    if (count != 51000) {
      throw new AssertionError("rows=" + count);
    }
    System.out.println(args[0] + ": verified " + count + " rows and both column values");
  }
}

Raw result lines

Before, direct reader:

Exception in thread "main" java.lang.AssertionError: row=40001 key=value_40001 value=value_40001
	at ReadDirectCodec.main(ReadDirectCodec.java:24)

Before, normal reader:

heap: verified 51000 rows and both column values

After, direct reader:

direct: verified 51000 rows and both column values

After, normal reader:

heap: verified 51000 rows and both column values

Also reproduced the failure using a file written by the current writer. The regression covers Snappy and Zstd, both page versions, dictionary and plain encoding, and the normal-reader control. It checks retained dictionary contents and uses tracking allocators to check buffer release.

Are there any user-facing changes?

Direct-codec reads preserve the stored column values. There is no file-format or public API change.