wzx140 opened a new issue, #10671:
URL: https://github.com/apache/paimon/issues/10671
# Paimon Partial-Update Validity Column Design
## Motivation
Ordinary partial update on primary-key tables uses NULL to mean "do not
update" and cannot distinguish an explicitly supplied NULL from a NULL inserted
to fill a missing column. This proposal explicitly declares an ordinary column
`_VALIDITY_FIELDS` in the table schema, using a bitmap to record the column
indexes participating in the update for each row. This column is visible to
users and included in `SELECT *`.
Assume an existing row `(1, A, B, C)`:
```sql
-- Update only b; the result is (1, A, B2, C).
INSERT INTO t (id, b) VALUES (1, 'B2');
-- Explicitly set a and c to NULL; the result is (1, NULL, B2, NULL).
INSERT INTO t (id, a, b, c) VALUES (1, NULL, 'B2', NULL);
```
The corresponding incremental records have the following storage layout. The
value column positions are `_VALIDITY_FIELDS=0, id=1, a=2, b=3, c=4`; the
bitmap does not mark column 0:
```text
[_KEY_id, _SEQUENCE_NUMBER, _VALUE_KIND, _VALIDITY_FIELDS, id, a, b, c]
Update only b:
[1, 100, INSERT, serialize({1, 3}), 1, NULL, B2, NULL]
Update a, b, and c, explicitly setting a and c to NULL:
[1, 101, INSERT, serialize({1, 2, 3, 4}), 1, NULL, B2, NULL]
```
## Public Interfaces
Add `merge-engine=partial-update-validity` (tentative name). Explicitly
declare `_VALIDITY_FIELDS VARBINARY` in the table.
```sql
CREATE TABLE t (
_VALIDITY_FIELDS VARBINARY,
id INT,
a STRING,
b STRING,
c STRING,
PRIMARY KEY (id) NOT ENFORCED
) WITH ('merge-engine' = 'partial-update-validity');
```
## Proposed Changes
### Core
#### Row and Storage Layout
`_VALIDITY_FIELDS` is an ordinary column in the complete value RowType. The
merge function reads its binary value from `kv.value()` by column position and
decodes it into an in-memory `RoaringBitmap32`.
```text
value schema: [_VALIDITY_FIELDS, id, a, b]
File layout:
[_KEY_<pk fields>, _SEQUENCE_NUMBER, _VALUE_KIND, <value fields>]
Example:
[1, 100, INSERT, serialize({1, 2}), 1, NULL, NULL]
```
The column uses ordinary `VARBINARY` encoding and is stored as a separate
Parquet `BYTE_ARRAY` column. It is copied, spilled, and serialized together
with the value.
Bitmap indexes correspond to target column positions in the complete value
schema; `_VALIDITY_FIELDS` does not mark itself. The merge function handles
this column separately and outputs the merged bitmap rather than applying
ordinary per-column overwrite logic to it.
#### Projection and Schema Evolution
Indexes in `validityFields` correspond to the complete business value schema
used when the file was written. Before the reader outputs an `InternalRow`, the
value and bitmap must both be mapped to the actual read schema.
`KeyValueFileReaderFactory` / `FormatReaderMapping` construct a position
mapping using stable field IDs from the file's business schema and the actual
read schema. Decode the bitmap from the ordinary column, transform it using the
mapping, and encode the result back into the corresponding value column:
```java
// mapping[newPos] = oldPos; -1 if the old column does not exist.
projectValidity(oldValid, mapping) {
newValid = new RoaringBitmap32();
for (newPos = 0; newPos < mapping.length; newPos++) {
oldPos = mapping[newPos];
if (oldPos >= 0 && oldValid.contains(oldPos)) {
newValid.add(newPos);
}
}
return newValid;
}
```
#### Read and Merge
The new merge engine is implemented by `ValidityPartialUpdateMergeFunction`,
which extends `PartialUpdateMergeFunction` to merge validity bits and preserve
the result. Merge each column from older to newer records in sequence order:
| Previous record valid | Incoming record valid | Result value | Result
record valid |
| --- | --- | --- | --- |
| 0 | 0 | Incoming value | 0 |
| 0 | 1 | Incoming value | 1 |
| 1 | 0 | Previous value | 1 |
| 1 | 1 | Incoming value | 1 |
```java
class ValidityPartialUpdateMergeFunction extends PartialUpdateMergeFunction {
// Merge records for the same primary key from older to newer sequence
numbers;
// delete and advanced aggregation branches are omitted.
reset() {
result = allNullRow();
resultValid = new RoaringBitmap32();
}
add(kv) {
row = kv.value();
valid = decodeOrNull(row.getBinary(validityColumnPos));
for (i = 0; i < row.getFieldCount(); i++) {
if (i == validityColumnPos) continue;
incomingValid = valid == null ? !row.isNullAt(i) :
valid.contains(i);
if (incomingValid || !resultValid.contains(i)) {
result[i] = valueAt(row, i); // Preserve values carried by
invalid columns; NULL can also be assigned.
}
if (incomingValid) {
resultValid.add(i);
}
}
}
getResult() {
result[validityColumnPos] = resultValid.serialize();
return KeyValue(..., result);
}
}
```
#### DEFAULT
Current behavior: DEFAULT converts NULL to the column default only during
writes. Reads and merges do not fill in defaults.
For example, the following SQL illustrates a table using `partial-update`,
where a has default D and omitted columns are filled with NULL by the upper
layer:
```sql
-- Schema: (id INT, a STRING DEFAULT 'D', b STRING), with id as the primary
key.
INSERT INTO t (id, a, b) VALUES (1, 'A', 'B');
SELECT id, a, b FROM t;
-- result: (1, 'A', 'B')
-- Only b is intended to change, but the filled-in a=NULL becomes D during
writing.
INSERT INTO t (id, b) VALUES (1, 'B2');
SELECT id, a, b FROM t;
-- result: (1, 'D', 'B2'); the previous value A is overwritten.
```
With `_VALIDITY_FIELDS`, NULL values are converted to DEFAULT according to
column configuration during writes for both `valid=0` and `valid=1`; this
conversion does not change the validity bits. During merging, valid values take
precedence over invalid values; when both the previous and incoming records are
invalid, the incoming value is used.
Using the new merge implementation `ValidityPartialUpdateMergeFunction`,
execute the same SQL on an empty table:
```sql
-- Schema: (id INT, a STRING DEFAULT 'D', b STRING), with id as the primary
key.
INSERT INTO t (id, a, b) VALUES (1, 'A', 'B');
SELECT id, a, b FROM t;
-- result: (1, 'A', 'B'); a=(A, valid=1).
-- Omit a: the filled-in NULL still becomes D, but a has valid=0.
INSERT INTO t (id, b) VALUES (1, 'B2');
SELECT id, a, b FROM t;
-- result: (1, 'A', 'B2'); the previous a is valid, so A is preserved rather
than overwritten by D.
```
### Lookup Changelog and Deletion Vectors
DV and changelog are not supported for now.
### Spark SQL
Add an analysis rule for INSERT / UPDATE / MERGE using the new merge engine.
Before filling missing columns and aligning assignments, preserve the actually
supplied columns, map them to column indexes in the complete table schema, and
automatically generate and populate the ordinary column `_VALIDITY_FIELDS`.
This column is the first column of the output row, and the bitmap does not mark
itself. MERGE generates the bitmap according to the action actually matched.
### PyPaimon
In `TableWrite.with_write_type(write_cols)`, map the business columns
supplied by the caller to column indexes in the complete table schema, generate
the validity bitmap, and automatically add `_VALIDITY_FIELDS` as the first
column in the internal write column list. Subsequent writes follow this plan to
automatically populate the encoded Arrow binary column, fill missing business
columns with NULL, and pass the complete row to Python's write and merge paths.
Callers do not need to manually add this column to write_cols or the input data.
## Compatibility and Migration
### Old Versions Reading New Files
Old Paimon clients do not recognize the new merge-engine value and fail when
opening the table with its configuration. A generic Parquet reader can still
read `_VALIDITY_FIELDS` as an ordinary binary column, but does not perform
validity-based merging.
### New Versions Reading Old Files
After the ordinary column is added to an existing table through schema
evolution, the reader fills the missing `_VALIDITY_FIELDS` in old files with
NULL. The merge function then follows the old semantics: non-NULL columns
participate in updates, while NULL columns do not. When reading old and new
files together, column positions are still mapped according to each file's
schema.
--
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]
