Variant shredding#

The standard Variant binary encoding is efficient and compact, but it must still read and decode each value for every query. The Parquet Variant shredding specification describes an optional on-disk layout for Variant columns that stores some fields in typed Parquet columns. Queries on the shredded fields read the typed columns directly, which is significantly faster.

Writing shredded Parquet columns#

Top-level fields#

Given the following table:

CREATE TABLE events AS
SELECT parse_json(json_data) AS event
FROM (VALUES
    ('{"service":"api","status":200,"latency_ms":12,"tags":["cached"]}'),
    ('{"service":"api","status":500,"latency_ms":85,"error":null}'),
    ('{"service":"api","status":504,"error":"Somethings not working"}')
) AS input(json_data);

You can store the service and status fields in typed columns by providing a variant.shredding option to the COPY statement:

COPY events TO 'shredded_events.parquet' STORED AS PARQUET
OPTIONS ('variant.shredding' '{"event": "STRUCT<service VARCHAR, status BIGINT>"}');

The variant.shredding option is a JSON object that maps the name of the Variant column to a SQL type. The full format is described below.

Storing the shredded column is logically equivalent to storing the typed service and status values plus a residual Variant column with the remaining fields:

+-------------------------------------+---------------+--------------+
| event (residual)                    | event.service | event.status |
+-------------------------------------+---------------+--------------+
| {"latency_ms":12,"tags":["cached"]} | api           | 200          |
| {"error":null,"latency_ms":85}      | api           | 500          |
| {"error":"Somethings not working"}  | api           | 504          |
+-------------------------------------+---------------+--------------+

Note

The actual encoding is more complicated: the typed columns and the binary residual are stored together in a single struct, as described in the Parquet Variant shredding specification.

Nested fields#

You can also shred nested objects and arrays. For example, if you have a requests table with a nested detail object and tags array:

CREATE TABLE requests AS
SELECT parse_json(json_data) AS request
FROM (VALUES
    ('{"service":"api","detail":{"latency_ms":12,"region":"us-east"},"tags":["cached"]}'),
    ('{"service":"api","detail":{"latency_ms":85},"tags":[]}'),
    ('{"service":"web","detail":{"region":"eu-west"},"error":"timeout"}')
) AS input(json_data);

To shred the latency_ms field of the nested detail object and the tags array, use the following COPY statement:

COPY requests TO 'shredded_requests.parquet' STORED AS PARQUET
OPTIONS ('variant.shredding' '{
  "request": "STRUCT<detail STRUCT<latency_ms BIGINT>, tags VARCHAR[]>"
}');

Fields that are not in the layout stay in the residual of the object that contains them: service in the residual of request, and region in the residual of request.detail. Rows without a shredded field have NULL in its typed column:

+-------------------------------------+---------------------------+---------------------------+--------------+
| request (residual)                  | request.detail (residual) | request.detail.latency_ms | request.tags |
+-------------------------------------+---------------------------+---------------------------+--------------+
| {"service":"api"}                   | {"region":"us-east"}      | 12                        | ["cached"]   |
| {"service":"api"}                   | NULL                      | 85                        | []           |
| {"error":"timeout","service":"web"} | {"region":"eu-west"}      | NULL                      | NULL         |
+-------------------------------------+---------------------------+---------------------------+--------------+

Format of the `variant.shredding` option#

variant.shredding is parsed as JSON, and its value must be a JSON object.

Each key is the case-sensitive name of a top-level Variant column in the COPY input that will be shredded. The value of each field is the SQL type used to store the shredded column.

The option is rejected for formats other than Parquet, and partition columns cannot be shredded.

One statement can shred several Variant columns. For example, given a table with two Variant columns, event and metric:

CREATE TABLE metrics AS
SELECT parse_json(event) AS event, parse_json(metric) AS metric
FROM (VALUES
    ('{"service":"api","status":200}', '42'),
    ('{"service":"api","status":500}', '7'),
    ('{"service":"web","status":504}', '"n/a"')
) AS input(event, metric);

The following COPY statement shreds the status field of event and the whole metric column:

COPY metrics TO 'shredded_metrics.parquet' STORED AS PARQUET
OPTIONS ('variant.shredding' '{
  "event":  "STRUCT<status BIGINT>",
  "metric": "BIGINT"
}');

The shredding is described as SQL types, which are parsed by DataFusion's SQL parser. The COPY statement attempts to "shred" the matching parts of the Variant into values of that Parquet type. If the Variant value can be converted to the specified type, the shredded column is populated with that value. Otherwise, a NULL is stored in the shredded column and the value remains in the residual Variant column.

For example, if you try to shred this Variant object as a BIGINT:

{
  "service": "api",
  "status": "200"
}

the shredded column will be NULL:

+----------------------------------+-----------------+
| residual Variant                 | shredded BIGINT |
+----------------------------------+-----------------+
| {"service":"api","status":"200"} | NULL            |
+----------------------------------+-----------------+

If a Variant object is shredded into a STRUCT, fields in the Variant that are not in the STRUCT remain in the residual Variant column. For example, if you shred a Variant object like

{
  "a": 1,
  "b": 2,
  "c": 3
}

into a STRUCT with only fields a and b (STRUCT<a BIGINT, b BIGINT>), the result is:

+------------------+-----------------+
| residual Variant | shredded STRUCT |
+------------------+-----------------+
| {"c":3}          | {a: 1, b: 2}    |
+------------------+-----------------+

Files with different shredding layouts#

It is also possible to read files with different shredding layouts as a single table, which is often referred to as schema evolution. This can happen when the optimal shredding evolves over time, or different producers shred the same data differently. DataFusion transparently unifies the different shredding layouts into a single table.

For example, if you write file1.parquet with the id and status fields shredded, and file2.parquet with just the id field shredded:

-- file1.parquet: shredded on id and status
COPY (
  SELECT parse_json(event) AS event FROM (VALUES
      ('{"id": 5, "service":"api", "status":200}'),
      ('{"id": 6, "service":"api", "status":500}'),
      ('{"id": 7, "service":"web", "status":504}')
  ) AS input(event)
) TO 'mixed_events/file1.parquet' STORED AS PARQUET
OPTIONS (
  'variant.shredding'
  '{"event": "STRUCT<id BIGINT, status BIGINT>"}' -- shred id and status
);

-- file2.parquet: shredded on id
COPY (
  SELECT parse_json(event) AS event FROM (VALUES
      ('{"id": 8, "service":"api", "status":200}'),
      ('{"id": 9, "service":"db", "status":503}')
  ) AS input(event)
) TO 'mixed_events/file2.parquet' STORED AS PARQUET
OPTIONS (
  'variant.shredding'
  '{"event": "STRUCT<id BIGINT>"}' -- shred id
);

Create a table over the directory, and DataFusion reads event from both files as a single column:

CREATE EXTERNAL TABLE mixed_events STORED AS PARQUET LOCATION 'mixed_events/';

SELECT event:id, event FROM mixed_events ORDER BY event:id::BIGINT;

+-----------------------+---------------------------------------+
| mixed_events.event:id | event                                 |
+-----------------------+---------------------------------------+
| 5                     | {"id":5,"service":"api","status":200} |
| 6                     | {"id":6,"service":"api","status":500} |
| 7                     | {"id":7,"service":"web","status":504} |
| 8                     | {"id":8,"service":"api","status":200} |
| 9                     | {"id":9,"service":"db","status":503}  |
+-----------------------+---------------------------------------+

Use CREATE EXTERNAL TABLE for such directories. Reading the directory directly with SELECT ... FROM 'mixed_events/' does not unify the layouts yet and returns an error or NULL for the fields that are shredded differently (apache/datafusion#27089).

See the documentation on VariantExprAdapterFactory for more details.

Supported types#

You can use any SQL type supported by DataFusion and listed in the shredding specification. For example:

BOOLEAN
TINYINT, SMALLINT, INT, BIGINT
FLOAT, DOUBLE, DECIMAL(p, s)
DATE, TIMESTAMP, TIMESTAMP WITH TIME ZONE (see below)
VARCHAR, BYTEA
STRUCT<name type, ...> and type[], nested to any depth

The following are rejected:

Unsigned integers, intervals, and other types without a shredded Parquet representation
TIME, because DataFusion's TIME has nanosecond precision and Variant times are microseconds
VARIANT fields inside a layout, which are not supported yet

To keep a field as Variant, leave it out of the layout.

TIMESTAMP WITH TIME ZONE uses the session time zone, datafusion.execution.time_zone, which is unset by default. Without it, the layout is timezone-naive like TIMESTAMP, and timezone-aware Variant timestamps stay in the residual. Variant stores these timestamps in UTC, so any zone works: run SET datafusion.execution.time_zone = 'UTC' before the COPY.

Variant shredding#

Writing shredded Parquet columns#

Top-level fields#

Nested fields#

Format of the variant.shredding option#

Files with different shredding layouts#

Supported types#

Format of the `variant.shredding` option#