JingsongLi commented on PR #8827:
URL: https://github.com/apache/paimon/pull/8827#issuecomment-5066373687
# Adaptive Cross-File Inference for Variant Shredding
## 1. Summary
Paimon currently infers a Variant shredding schema independently for every
data file. Each file
buffers up to `variant.shredding.maxInferBufferRow` rows, infers its
physical schema, flushes the
buffer, and then continues writing. This approach is robust to schema
changes, but it repeats the
same inference work for adjacent files and may produce unnecessary
physical-schema churn.
This document proposes an adaptive inference mode with three properties:
1. **Previous-file prior:** retain bounded inference evidence and the
selected physical schema from
the preceding file in the same rolling writer.
2. **Small-sample correction:** inspect a smaller prefix of the next file
before finalizing its
physical schema.
3. **Hysteresis:** use different thresholds for retaining an existing
shredded path and admitting a
new path, reducing oscillation near the cardinality threshold.
The prior is an optimization hint, not a correctness dependency. Every file
remains self-describing,
and values that do not fit a selected typed layout continue to be stored in
the Variant `value`
field according to the Parquet Variant shredding model.
The state is intentionally scoped to one `RollingFileWriter`. It is not
shared across tasks,
processes, partitions, buckets, or writer restarts, and it is not persisted
in table metadata.
## 2. Motivation
Independent per-file inference has three costs.
### 2.1 Repeated buffering
For a stable input stream, adjacent files often infer almost identical
schemas. Buffering the same
large prefix for every file adds memory pressure and delays creation of the
underlying format
writer. It can also cause a file to exceed its target size because rolling
cannot be evaluated until
the inferred plan has been finalized.
### 2.2 Physical-schema churn
A field whose observed cardinality is close to
`variant.shredding.minFieldCardinalityRatio` may be included in one file and
excluded from the next.
The data remains readable, but frequent layout changes reduce predictability
and can make projected
reads and data skipping less effective.
### 2.3 Weak inference for small files
Small files may close before enough rows have been observed. A bounded prior
supplies evidence from
the immediately preceding stream segment without requiring table-wide
coordination.
## 3. Goals
- Reduce inference buffering after the first file in a rolling writer.
- Preserve the ability to react to new fields and type changes at file
boundaries.
- Reduce repeated admission and eviction of fields near the selection
threshold.
- Preserve file-level correctness and independent readability.
- Keep memory usage bounded by configured schema width and a fixed effective
prior size.
- Make the result deterministic for the same prior and current sample.
- Avoid cross-writer synchronization and persistent table metadata in the
first version.
- Preserve the existing configured-schema behavior.
## 4. Non-Goals
- Producing one globally stable shredding schema for an entire table.
- Learning from files written by other tasks or writers.
- Persisting inference state across checkpoints, retries, or process
restarts.
- Selecting fields from query history.
- Changing the Variant encoding or Parquet shredding specification.
- Changing the current strict type-merging policy in the first
implementation.
- Reconstructing prior statistics by reading the previous file footer.
Query-aware pinned paths and table-level schema pinning are complementary
future features.
## 5. Current Behavior
`InferShreddingWritePlanWriter` buffers the prefix of each file and invokes
`ShreddingWritePlanFactory.createWritePlan` once the row limit is reached or
the writer closes.
`VariantShreddingWritePlanFactory` creates a new
`InferVariantShreddingSchema` and derives a
physical row type only from those buffered rows.
`RowDataRollingFileWriter` creates a new file-writer context for every
rolled file. That operation
also creates a new format writer factory and therefore a new
`VariantShreddingWritePlanFactory`.
No inference information survives the file boundary.
The current selection limits are:
- cold sample limit: 4,096 rows by default;
- minimum field cardinality ratio: 0.10;
- maximum inferred schema width: 300;
- maximum inference depth: 50.
## 6. Design Overview
Adaptive inference introduces one in-memory
`VariantShreddingInferenceSession` for each sequential
rolling writer.
```text
File 1 prefix (cold sample)
|
v
analyze + select plan
|
+----> bounded evidence + selected plan
|
File 2 prefix (warm sample) |
| |
+---------------------+
|
v
combine evidence + apply hysteresis + correct types
|
+----> File 2 plan and updated bounded prior
```
The first file uses the existing cold-sample limit. Later files use a
smaller warm-sample limit.
For every warm file, the session:
1. analyzes the current prefix;
2. combines current field-presence evidence with bounded prior evidence;
3. applies admission and retention thresholds;
4. validates the previous typed layout against current observations;
5. applies the existing depth and width limits;
6. creates the per-file write plan; and
7. replaces the session state with a bounded posterior for the next file.
## 7. State Scope and Lifecycle
### 7.1 Scope
The session belongs to a single `RollingFileWriter` instance. In normal
Paimon write paths, that
also limits it to one logical stream such as a partition/bucket/level writer.
The session must not be stored in:
- the shared `FileFormat`, which is required to be thread-safe;
- static state;
- a table-wide cache; or
- a cache keyed only by table or column name.
There is no globally meaningful "previous file" when multiple tasks write
concurrently.
### 7.2 Lifecycle
- The session is created with the rolling writer.
- The first inferred plan initializes the session.
- A subsequent file reads and updates the session only from the writer
thread.
- The rolling writer opens the next file only after the current file closes
successfully.
- If writing or closing fails, the rolling writer aborts and discards the
entire session.
- A retry creates a new session and starts with cold inference.
- A configured shredding schema takes precedence and bypasses adaptive
inference.
No extra commit protocol is required because the state is ephemeral and
cannot be observed by
readers.
## 8. Inference Evidence
The implementation should separate **analysis** from **schema
materialization**. The current
`InferVariantShreddingSchema` combines both operations, which makes it
difficult to reuse bounded
evidence without carrying a partially materialized `RowType`.
A conceptual model is:
```java
final class VariantShreddingInferenceState {
private final Map<List<Integer>, VariantColumnEvidence> columns;
private final RowType previousPhysicalRowType;
private final long completedFileCount;
}
final class VariantColumnEvidence {
private final double effectiveRootValueCount;
private final Map<VariantFieldPath, FieldEvidence> fields;
}
final class FieldEvidence {
private final double effectivePresenceCount;
private final DataType previousSelectedType;
private final boolean previouslySelected;
}
```
The actual implementation may use compact tree nodes instead of path-keyed
maps. Field paths and
types must use deterministic equality and ordering.
For the first version, the reusable evidence only needs:
- the effective number of non-null root Variant values;
- the effective presence count of each object field;
- whether the path was selected in the preceding plan; and
- the type selected in the preceding plan.
The current sample still supplies exact type observations. Persisting a
complete type histogram or
full-file fallback statistics can be added later.
### 8.1 Cardinality denominator
To preserve current behavior, a nested field's cardinality remains relative
to the number of
non-null values of its root Variant column, rather than only to the number
of occurrences of its
immediate parent.
Each Variant column maintains its own denominator. If the current prefix
contains no non-null value
for a Variant column, that column's prior is retained without decay because
the sample provides no
evidence about its shape.
## 9. Combining Prior and Current Evidence
Raw counts from the cold sample must not dominate all future files. The
prior is therefore converted
to a bounded effective sample.
Let:
- `P` be the previous bounded evidence;
- `C` be evidence from the current warm prefix; and
- `W` be the warm-sample row limit.
For each Variant column:
1. scale `P` so that its effective root count is at most `W`;
2. add the current counts from `C`;
3. compute field presence ratios from the combined counts; and
4. scale the combined evidence back to at most `W` before storing it as the
next prior.
Conceptually:
```text
boundedPrior = scaleToAtMost(previousEvidence, W)
posterior = boundedPrior + currentEvidence
nextPrior = scaleToAtMost(posterior, W)
```
With a full warm sample and the default equal cap, the previous evidence and
the current file have
approximately equal influence. Repeated absence naturally decays a field's
rate: 100%, 50%, 25%,
12.5%, and so on. The prior can therefore stabilize a path without making it
permanent.
The first file may analyze up to the existing 4,096-row limit, but its
evidence is scaled down before
it becomes a prior.
## 10. Hysteresis and Field Selection
Let:
- `A` be the existing admission threshold,
`variant.shredding.minFieldCardinalityRatio`; and
- `R` be the retention threshold, with `0 <= R < A`.
The proposed default is:
```text
A = 0.10
R = 0.05
```
A field is eligible when:
```text
previously selected: combinedPresenceRatio >= R
not previously selected: combinedPresenceRatio >= A
```
This gap prevents a field near 10% from alternating between shredded and
unshredded layouts on
every file.
### 10.1 Width-budget ordering
Hysteresis must not allow stale fields to permanently occupy a full width
budget. Eligible fields
are therefore ranked by:
1. combined presence ratio, descending;
2. previously selected status, selected first when ratios are equal; and
3. complete field path, lexicographically ascending.
Parent nodes required to reach a selected child are included before the
child. Existing width and
depth accounting remains authoritative. If two schemas consume the same
budget, the deterministic
ordering produces the same result for the same evidence.
## 11. Type Selection and Correction
Field-membership hysteresis must not make type changes unnecessarily sticky.
For a path selected in the previous plan:
1. If the current sample has no value for the path, retain the previous
selected type.
2. If all current observations are compatible with the previous type, retain
it and apply existing
decimal/integer widening rules when required.
3. If any current observation is incompatible, discard the previous type
hint and infer the type
from the current sample using the existing strict merge rules.
For a newly admitted path, infer its type from the current sample using the
existing rules.
This policy deliberately applies hysteresis to **field membership**, but
immediate correction to
**incompatible types**. It avoids carrying a stale typed layout across
several files while keeping
the first implementation behaviorally close to current Paimon.
Values after the sampled prefix may still fail to fit the selected type.
They remain lossless in the
Variant `value` field. This is a correctness fallback, but a high fallback
rate can reduce
compression, projection, and data-skipping benefits.
A follow-up design may select the most common compatible type with a
configurable fit threshold,
similar to other Variant implementations. That policy change should be
evaluated separately.
## 12. Buffer Limits
The adaptive mode uses:
- the existing `variant.shredding.maxInferBufferRow` for the first file; and
- a smaller warm-file row limit for subsequent files.
Inference should also stop at a byte limit. A row-only limit is insufficient
when a small number of
Variant values contain large strings, objects, arrays, or binary values.
The writer should finalize its plan when either limit is reached:
```text
bufferedRows >= rowLimit || estimatedBufferedBytes >= byteLimit
```
The byte estimator does not need to model Java object overhead exactly. It
should conservatively
include Variant binary/value bytes and fixed row overhead, be deterministic,
and avoid traversing
the same Variant tree twice.
## 13. Writer Integration
The smallest integration is to create the format writer factory once per
`RowDataRollingFileWriter`, then reuse that factory for the sequential files
produced by the rolling
writer.
Today, `RowDataRollingFileWriter` calls:
```text
fileFormat.createWriterFactory(writeSchema)
```
inside the per-file supplier. In adaptive mode, it should happen once when
the supplier is built.
For Parquet, the resulting `ShreddingWritePlanWriterFactory` contains one
`VariantShreddingWritePlanFactory`, which in turn owns the adaptive session.
Per-file components must remain per-file:
- output streams;
- `RowDataFileWriter`;
- statistics producers and collectors;
- file-index writers;
- sidecar writers; and
- writer metadata.
Only the reusable `FormatWriterFactory` and its bounded inference session
cross file boundaries.
`ShreddingWritePlanFactory.inferBufferRowCount()` may return the cold or
warm limit based on whether
the session has a prior. `createWritePlan(sampleRows)` analyzes the current
sample, materializes the
plan, and advances the session.
The factory is used sequentially by one rolling writer. It is not advertised
as thread-safe and
must not be shared between rolling writers.
## 14. Configuration
The following options are proposed:
| Option | Default | Description |
| --- | ---: | --- |
| `variant.shredding.inferenceMode` | `PER_FILE` | `PER_FILE` preserves
current behavior; `ADAPTIVE` enables rolling-writer-scoped prior reuse. |
| `variant.shredding.adaptive.maxInferBufferRow` | `256` | Maximum prefix
rows for files after the first file in the session. |
| `variant.shredding.adaptive.retentionRatio` | `0.05` | Minimum combined
presence ratio for a path selected in the previous plan. Must not exceed
`variant.shredding.minFieldCardinalityRatio`. |
| `variant.shredding.maxInferBufferBytes` | `64 MB` | Maximum estimated
bytes retained for inference in either mode. |
The existing options continue to define:
- cold-file row limit;
- admission threshold;
- maximum schema width; and
- maximum schema depth.
The initial release should keep `PER_FILE` as the default. After production
evidence demonstrates
stable write and read improvements, changing the default can be considered
independently.
## 15. Correctness and Compatibility
### 15.1 Read correctness
Each file persists its own physical Variant shredding schema. Readers
already support different
physical layouts across files. The adaptive session affects only how a file
chooses that layout.
Values incompatible with a typed field are encoded in `value`, so prior
reuse does not discard or
coerce user data.
### 15.2 Data skipping
A stale type may increase non-null `value` occurrences. Typed-column
statistics alone cannot
describe those fallback values. The adaptive design therefore uses the
current sample to correct
incompatible prior types immediately.
Future full-file fallback metrics should be used to detect paths whose typed
layout is not
beneficial even when the prefix appeared compatible.
### 15.3 Backward compatibility
- No file-format or table-metadata change is required.
- Existing files require no rewrite.
- Configured shredding schemas are unchanged.
- `PER_FILE` mode preserves the current inference lifecycle.
- Adaptive and non-adaptive files may coexist in the same table.
## 16. Failure, Retry, and Concurrency Semantics
- Session mutations occur only in the rolling writer's thread.
- The next file cannot consume the updated prior until the current file has
closed.
- If the current file fails, the writer aborts and no subsequent file
observes its prior.
- A task retry starts cold; its physical layouts may differ from the failed
attempt, which is valid
because layouts are file-local.
- Concurrent writers never share inference sessions.
- Schema evolution creates a new rolling writer/session. If an
implementation can reuse a writer
across logical schema changes, it must reset the session when the
logical-row-type fingerprint
changes.
## 17. Observability
Adaptive mode should expose metrics or debug counters for:
- cold versus warm inference count;
- sampled rows and estimated bytes;
- inference time;
- number of admitted, retained, evicted, and type-corrected paths;
- inferred physical-schema width and depth;
- physical-schema fingerprint changes between adjacent files; and
- Variant values or bytes falling back from `typed_value` to `value`, if the
writer can report them.
The schema fingerprint should be deterministic and suitable for measuring
churn, but it does not
need to be persisted in table metadata.
## 18. Test Plan
### 18.1 Unit tests
Add focused tests for:
- first-file cold inference and later-file warm limits;
- stable adjacent files retaining the same plan;
- a new hot field being admitted;
- an old field decaying below the retention threshold;
- a field oscillating between the admission and retention thresholds;
- a full width budget replacing a stale low-frequency field with a hotter
new field;
- immediate correction of an incompatible previous type;
- numeric widening compatible with the previous type;
- no current non-null Variant values retaining the prior without decay;
- independent state for multiple Variant columns;
- nested fields, arrays, nulls, and depth limits;
- deterministic ordering for equal-frequency paths; and
- state reset after creating a new session.
### 18.2 Writer integration tests
Use a small target file size to force one rolling writer to create multiple
Parquet files, then
verify:
- the first and warm-file sample limits;
- the physical schema stored in each file;
- logical read equality across mixed physical layouts;
- lossless fallback for values that appear after the sampled prefix;
- behavior when a file fails or is aborted; and
- no state leakage between two rolling writers.
### 18.3 Benchmarks
Compare:
1. current per-file inference with a 4,096-row sample;
2. direct previous-plan reuse without correction;
3. adaptive prior plus a 256-row correction sample; and
4. configured/pinned schema.
Use the following workloads:
- stationary schemas;
- sparse Zipf-distributed keys;
- fields near the cardinality threshold;
- abrupt schema shifts at file boundaries;
- alternating types;
- large Variant values;
- partition- or sensor-sorted streams; and
- compaction output mixing several input schemas.
Measure:
- write throughput and time to initialize each physical writer;
- peak inference-buffer bytes;
- schema-fingerprint churn;
- shredded-path coverage and fallback bytes;
- file size and Parquet leaf-column count;
- projected read bytes and latency; and
- predicate/data-skipping effectiveness.
## 19. Rollout Plan
1. Refactor Variant inference into analysis, evidence combination, and
schema materialization
without changing current behavior.
2. Add the byte cap to per-file inference.
3. Add the rolling-writer-scoped session behind `ADAPTIVE` mode.
4. Add unit, integration, and benchmark coverage.
5. Run shadow metrics that compare the adaptive plan with the current
per-file plan without changing
the written schema.
6. Enable adaptive mode in selected workloads and evaluate schema churn,
fallback rate, memory, and
read performance.
7. Consider changing the default only after results are stable across
stationary and schema-shift
workloads.
## 20. Alternatives Considered
### 20.1 Reuse the previous physical schema without sampling
This minimizes buffering but cannot detect new hot paths or boundary-aligned
schema changes. It may
increase fallback values and weaken data skipping. It is not recommended.
### 20.2 Use a smaller independent sample for every file
This reduces buffering but loses evidence for stable sparse fields and
increases threshold noise.
It does not address physical-schema churn.
### 20.3 Read the previous file footer
The footer exposes the selected physical schema but not the observations
that caused the selection,
including frequency and rejected alternatives. The writer already has richer
information in memory,
so footer I/O adds latency without improving the prior.
### 20.4 Persist one table-wide learned schema
This can provide stronger stability but requires coordination, evolution
rules, ownership, and
metadata compatibility. It is better treated as explicit schema pinning
rather than an implicit
extension of per-file inference.
### 20.5 Aggregate compaction input schemas
This is useful for compaction because there is no single representative
previous input file.
Compaction-aware aggregation is compatible with this design but should be a
separate extension.
## 21. Open Questions
- Should the warm-sample default be 128, 256, or derived from the target
file size?
- Is a 0.05 absolute retention threshold preferable to a fixed fraction of
the admission threshold?
- Can the Variant writer report full-file fallback counts cheaply enough to
feed the next prior?
- Should compaction initialize its session from aggregated input-file
metadata?
- Which paths, if any, should users be able to pin ahead of inferred paths?
- After the adaptive lifecycle is proven, should type selection move from
strict consistency to a
minimum fit ratio with minority values stored in `value`?
## 22. References
- [Parquet Variant
Shredding](https://github.com/apache/parquet-format/blob/master/VariantShredding.md)
- [Databricks Variant
Shredding](https://docs.databricks.com/aws/en/tables/features/variant-shredding)
- [Delta Lake Variant Shredding
Protocol](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#variant-shredding)
- [Apache Spark Variant Shredding
Inference](https://github.com/apache/spark/pull/52406)
- [Apache Iceberg Variant Shredding
Analyzer](https://github.com/apache/iceberg/blob/main/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java)
- [DuckDB Automatic Variant
Shredding](https://github.com/duckdb/duckdb/pull/19336)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
