Apache ORC Documentation

Apache ORC (Optimized Row Columnar) is a self-describing, type-aware columnar file format designed for Hadoop workloads. It is optimized for large streaming reads and supports predicate pushdown, allowing readers to use internal indexes to quickly find required rows or stripes.

Key features:

• Columnar Storage: Enables reading, decompressing, and processing only the specific columns required for a query.
• Type Awareness: The writer automatically selects the most appropriate encoding for each data type.
• Complex Type Support: Supports the complete set of Hive types, including structs, lists, maps, and unions.
• Indexing: Built-in indexes allow for efficient row-level searches (narrowing down to sets of 10,000 rows).

ORC 2.1.0 includes several significant updates:

C++ Features

• Async Prefetch: Support for async prefetch in the Orc reader.
• Schema Evolution: Expanded support for evolving types (decimal, timestamp, string, numeric).
• Performance: Improved writing performance for encoded string columns.

Java & Build

• Hadoop Upgrade: Upgraded to Hadoop 3.4.1.
• Dependency Updates: Significant bumps to guava, slf4j, zstd-jni, and protobuf-java.
• Tooling: Added merge command documentation for Java tools.

ORC provides two distinct timestamp types. The choice depends on whether you need the time to be relative to the reader's location or a fixed point in time.

• timestamp: Represents a date and time without a time zone. The value remains constant regardless of the reader's time zone (e.g., 10:00 remains 10:00 in both Los Angeles and New York).
• timestamp with local time zone: Represents a fixed instant in time. The displayed value changes based on the reader's time zone to reflect that specific instant.

Recommendation: Unless your application uses UTC consistently, timestamp with local time zone is strongly preferred for most use cases to ensure events are correctly interpreted as specific points in time.

The Core ORC API uses VectorizedRowBatch to pass data efficiently. A VectorizedRowBatch contains data for 1024 rows, optimized for speed by allowing direct field access. It consists of an array of ColumnVector objects (cols) and an integer size representing the number of rows in the batch.

To work with vectorized data, you must interact with the cols array and respect the size property to avoid accessing uninitialized rows.

package org.apache.hadoop.hive.ql.exec.vector;

public class VectorizedRowBatch {
  public ColumnVector[] cols;
  public int size;
  ...
}

ORC uses RLEv2 for integer compression, which selects one of four sub-encodings based on the data pattern to optimize compression and expansion speed:

• Short Repeat: For short sequences of repeated values. Minimizes header overhead.
• Direct: For random sequences with a relatively constant bit width. Uses fixed-width big endian encoding.
• Patched Base: For sequences with highly variable bit widths. It uses a base value and a 95th percentile bit width, applying 'patches' for the remaining 5% of values.
• Delta: For monotonically increasing or decreasing sequences. Encodes the first value and then the deltas between subsequent values.

To record positions in the RowIndex, the method depends on whether the stream is compressed:

• Uncompressed streams: The position is the byte offset of the RLE run's start location, followed by the number of values that need to be consumed from the run.
• Compressed streams: The position consists of the start of the compression chunk in the stream, the number of decompressed bytes that need to be consumed, and finally the number of values consumed in the RLE.

ORC files represent data as logically sequenced objects. While Hive typically uses a struct as the root object type to represent top-level columns, this is not a requirement of the ORC format.

Compound types create a tree structure of child columns. For example, a map will always have two child columns, and a union will have one child column for every possible variant.

Example schema structure:

create table Foobar (
 myInt int,
 myMap map<string, struct<myString : string, myDouble: double>>,
 myTime timestamp
);

In this example, the myMap column would branch into child columns for the map keys and values, and the value column would further branch into child columns for the struct fields (myString and myDouble).

When using generic compression codecs (like zlib or snappy), ORC writes data in independent chunks to allow readers to skip over compressed bytes without decompressing the entire stream.

Chunk Structure:

• Each chunk has a 3-byte header.
• The header contains a little-endian value: (compressedLength * 2 + isOriginal).
• If the compressed data is larger than the original, the isOriginal flag is set and the original uncompressed data is stored instead.
• The default chunk size is 256K, though writers can adjust this. Larger chunks improve compression ratios but increase memory requirements.
• The chunk size is recorded in the Postscript so readers can allocate appropriate buffers.

ORC supports static data masking, where user data is masked before the unencrypted variant is written to the file. This masking is purely informational for readers.

Standard masks include:

• nullify: All values become null (default).
• redact: Replaces characters with constants (e.g., X or 9).
• sha256: Replaces a string with its SHA-256 hash.

Constraints:

• Masks cannot change the data type of a column, only the values.
• Users may define custom masks.