GH-2870: Report compressed dictionary page sizes in the CLI

[1fanwang]

Rationale for this change

parquet pages reports decoded dictionary sizes instead of the bytes stored in compressed files. It can also fail on a valid file whose dictionary is not used by any data page. The summary now reports the stored page sizes for both layouts.

Closes #2870.

What changes are included in this PR?

Read dictionary sizes from the original page headers and start the scan at the column's physical offset. An unused dictionary no longer makes the scan start at the first data page. Files and the core page API are unchanged.

Are these changes tested?

Testing Done

Ran both fixtures with JDK 17.0.5 and Thrift 0.24.0 against the upstream baseline and this PR's head.

Scenario	Command	Before	After
GZIP dictionary	java -jar <cli.jar> pages -c color compressed-dictionary.parquet	248 B; 124.00 B per entry	33 B; 16.50 B per entry
Unused dictionary followed by plain data	java -jar <cli.jar> pages unused-dictionary.parquet	Required field 'uncompressed_page_size' was not found in serialized data!	Dictionary 12 B; data page 8 B
Decode the unused-dictionary file	java -jar <cli.jar> cat unused-dictionary.parquet	Values 41 and 42	Values 41 and 42

From this PR's checkout, build the two runtime jars:

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

Create and inspect the compressed file:

python3 -c 'from pathlib import Path; Path("dictionary_page_input.csv").write_text("color\n" + ("a" * 120 + "\n" + "b" * 120 + "\n") * 100)'
java -Xmx512m -XX:ActiveProcessorCount=2 -jar before-cli.jar convert-csv dictionary_page_input.csv --require color --compression-codec GZIP -o compressed-dictionary.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -jar before-cli.jar pages -c color compressed-dictionary.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -jar after-cli.jar pages -c color compressed-dictionary.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -jar after-cli.jar pages --raw -c color compressed-dictionary.parquet

Save the writer below as WriteUnusedDictionary.java, then create a file with a dictionary and only plain-encoded data:

java -Xmx512m -XX:ActiveProcessorCount=2 -cp before-cli.jar WriteUnusedDictionary.java unused-dictionary.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -jar before-cli.jar pages unused-dictionary.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -jar after-cli.jar pages unused-dictionary.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -jar after-cli.jar pages --raw unused-dictionary.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -jar before-cli.jar cat unused-dictionary.parquet
java -Xmx512m -XX:ActiveProcessorCount=2 -jar after-cli.jar cat unused-dictionary.parquet

Executed fixture writer

import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.bytes.BytesInput;
import org.apache.parquet.column.Encoding;
import org.apache.parquet.column.page.DictionaryPage;
import org.apache.parquet.column.statistics.Statistics;
import org.apache.parquet.hadoop.ParquetFileWriter;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.parquet.hadoop.util.HadoopOutputFile;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Types;

public class WriteUnusedDictionary {
  public static void main(String[] args) throws Exception {
    PrimitiveType type = Types.required(PrimitiveType.PrimitiveTypeName.INT32).named("value");
    MessageType schema = new MessageType("record", type);
    try (ParquetFileWriter writer = new ParquetFileWriter(
        HadoopOutputFile.fromPath(new Path(args[0]), new Configuration()), schema,
        ParquetFileWriter.Mode.CREATE, ParquetWriter.DEFAULT_BLOCK_SIZE,
        ParquetWriter.MAX_PADDING_SIZE_DEFAULT)) {
      writer.start();
      writer.startBlock(2);
      writer.startColumn(schema.getColumnDescription(new String[] {"value"}), 2, CompressionCodecName.UNCOMPRESSED);
      writer.writeDictionaryPage(new DictionaryPage(
          BytesInput.concat(BytesInput.fromInt(10), BytesInput.fromInt(20), BytesInput.fromInt(30)), 3, Encoding.PLAIN));
      writer.writeDataPage(2, 2 * Integer.BYTES,
          BytesInput.concat(BytesInput.fromInt(41), BytesInput.fromInt(42)),
          Statistics.createStats(type), Encoding.RLE, Encoding.RLE, Encoding.PLAIN);
      writer.endColumn();
      writer.endBlock();
      writer.end(Map.of());
    }
  }
}

Raw result lines

GZIP dictionary before:

  0-D    dict  G _  2       124.00 B   248 B     
  0-1    data  G R  200     0.13 B     25 B                        

GZIP dictionary after:

  0-D    dict  G _  2       16.50 B    33 B      
  0-1    data  G R  200     0.13 B     25 B                        

Unused-dictionary file before:

Unknown error
java.lang.RuntimeException: java.io.IOException: can not read class org.apache.parquet.format.PageHeader: Required field 'uncompressed_page_size' was not found in serialized data! Struct: org.apache.parquet.format.PageHeader$PageHeaderStandardScheme@2a65bb85

Unused-dictionary file after:

  0-D    dict  _ _  3       4.00 B     12 B      
  0-1    data  _ _  2       4.00 B     8 B                         

Decoded rows, unchanged:

{"value": 41}
{"value": 42}

The regression also verifies that the unused dictionary has a recorded offset but no dictionary-encoded data pages. Raw headers, decoded rows, and the original GZIP data-page summary remain unchanged.

Are there any user-facing changes?

Dictionary summaries report on-disk bytes, including when no data page uses the dictionary.

[divjotarora]

Seems reasonable, but one possible edge case

[divjotarora]

getPageCompressedSize calls columnChunk.hasDictionaryPage() to determine the page offset. This function does:

  public boolean hasDictionaryPage() {
    EncodingStats stats = getEncodingStats();
    if (stats != null) {
      // ensure there is a dictionary page and that it is used to encode data pages
      return stats.hasDictionaryPages() && stats.hasDictionaryEncodedPages();
    }

    Set<Encoding> encodings = getEncodings();
    return (encodings.contains(PLAIN_DICTIONARY) || encodings.contains(RLE_DICTIONARY));
  }

In an edge case where a writer emits a dictionary page followed by no PLAIN_DICTIONARY or RLE_DICTIONARY data pages, we would get wrong results because hasDictionaryPage() returns false.

Realistically, writers wouldn't do this, but I didn't find any wording in the spec explicitly disallowing it. The TestParquetFileWriter code in this repo seems to do exactly this.