Skip to content

[BP-2.4][FLINK-41180][core] Sort VARIANT object fields by UTF-8 bytes #29622

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
raminqaf merged 3 commits into from
Oct 27, 2026
Merged

[BP-2.4][FLINK-41180][core] Sort VARIANT object fields by UTF-8 bytes #29622

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2e3589c
[FLINK-41110][core] Compare UTF-8 bytes to find VARIANT object fields
raminqaf Oct 22, 2026
72f751e
[hotfix][docs] Sync the zh VARIANT data type docs with en
raminqaf Oct 26, 2026
2d0a444
[FLINK-41180][core] Sort VARIANT object fields by UTF-8 bytes
raminqaf Oct 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 43 additions & 7 deletions docs/content.zh/docs/sql/reference/data-types.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1695,6 +1695,8 @@ A scalar value can also be cast to a `VARIANT` with `CAST` or `TRY_CAST`. Only a
stored with microseconds, and above with nanoseconds, even when the value has no digits below a
microsecond. Nanoseconds only cover 1677-09-21 to 2262-04-11, so for a precision above 6 a value
outside that range fails the cast, and `TRY_CAST` returns `NULL`.
- A `TIME` holds the milliseconds of one day. Only a source or a function can produce a value
outside that range, but such a value fails the cast, and `TRY_CAST` returns `NULL`.
- A `NaN` or infinite `FLOAT` or `DOUBLE` is stored as is, although `PARSE_JSON` rejects them. A
`VARIANT` is not limited to JSON, but JSON has no literal for them. So `JSON_STRING` fails on such
a variant, and so does writing it to a sink with the `json` or `raw` format, which fails the job.
Expand All @@ -1707,7 +1709,7 @@ A scalar value can also be cast to a `VARIANT` with `CAST` or `TRY_CAST`. Only a
`CAST`.
- A constructed type casts element by element when its element, field, or value type casts to
`VARIANT`, for example `ARRAY<INT>` to `ARRAY<VARIANT>`. A `NULL` element stays a SQL `NULL`. A
whole `ARRAY`, `MAP`, or `ROW` does not cast into a single `VARIANT` yet.
whole constructed value can also become a single `VARIANT`, see below.
- As a `MAP` key or a `MULTISET` element, two `VARIANT` values match only when their binary
encodings match. So a `1` cast from `INT` does not match a `1` cast from `BIGINT` or parsed by
`PARSE_JSON('1')`.
Expand All @@ -1720,7 +1722,40 @@ CAST(NULL AS VARIANT) -- NULL
CAST(CAST('NaN' AS DOUBLE) AS VARIANT) -- NaN, stored as a DOUBLE
CAST(INTERVAL '2' DAY AS VARIANT) -- fails at validation
CAST(ARRAY[1, NULL] AS ARRAY<VARIANT>) -- [1, NULL], each element a VARIANT, the NULL stays SQL NULL
CAST(ARRAY[1, 2] AS VARIANT) -- fails at validation, not supported yet
```

A whole `ARRAY`, `MAP`, `ROW`, or `STRUCTURED` value can also be cast into a single `VARIANT`. An
`ARRAY` becomes a variant array, and a `MAP`, `ROW`, or `STRUCTURED` value a variant object. Each
leaf is stored by the rules above, so the cast is supported only when every leaf type casts to
`VARIANT`.

- A `ROW` or `STRUCTURED` value is keyed by its field names.
- A `MAP` needs a character string key, which becomes the object key. A `NULL` key fails the cast.
If a key appears twice, the last value is kept.
- A variant object sorts its keys by their UTF-8 bytes, so neither the field order of a `ROW` nor
the entry order of a `MAP` is kept.
- A `NULL` element, field, or map value becomes a variant null, so an array keeps its length and an
object keeps its keys. A cast element by element differs here: `ARRAY<INT>` to `ARRAY<VARIANT>`
keeps a SQL `NULL`, while `ARRAY<INT>` to `VARIANT` turns it into a variant null.
- A nested `VARIANT` is embedded as is.
- The whole value must fit into the 16 MiB of a `VARIANT`. Any `ARRAY` or `MAP` can exceed it, and so
can a `ROW` with a nested `VARIANT` or with fields whose declared sizes add up to more. The cast
then fails, and `TRY_CAST` returns `NULL` for the whole value.
- Casting the `VARIANT` back to the original type returns the original value, since a cast to `ROW`
matches fields by name. There are two exceptions:
- A SQL `NULL` in a `VARIANT` field, in an `ARRAY<VARIANT>`, or as a value of a
`MAP<STRING, VARIANT>` comes back as a variant null.
- A `MAP` that holds a key twice comes back with only the last value for that key.

The following examples use `r` for a column of type `ROW<name STRING, id INT>` that holds
`('ada', 7)`:

```sql
CAST(ARRAY[1, NULL] AS VARIANT) -- [1, null], the NULL becomes a variant null
CAST(MAP['b', 2, 'a', 1] AS VARIANT) -- {"a": 1, "b": 2}
CAST(r AS VARIANT) -- {"id": 7, "name": "ada"}
CAST(MAP[1, 'a'] AS VARIANT) -- fails at validation, a MAP key must be a character string
CAST(ARRAY[INTERVAL '1' DAY] AS VARIANT) -- fails at validation
```

A `VARIANT` has no equality and no order, because one value has many valid binary encodings. The
Expand Down Expand Up @@ -1949,15 +1984,15 @@ COALESCE(TRY_CAST('non-number' AS INT), 0) --- 结果返回数字 0 的 INT 格
| `FLOAT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N |
| `DOUBLE` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N |
| `DATE` | Y | N | N | N | N | N | N | N | N | N | Y | N | Y | Y | N | N | N | N | N | N | N | Y | N |
| `TIME` | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | N | N | Y | N |
| `TIME` | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | N | N | ! | N |
| `TIMESTAMP` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | ! | N |
| `TIMESTAMP_LTZ` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | ! | N |
| `INTERVAL` | Y | N | N | N | N | N | Y⁵ | Y⁶ | N | N | N | N | N | N | Y | N | N | N | N | N | N | N | N |
| `ARRAY` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | N | N |
| `ARRAY` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | !⁸ | N |
| `MULTISET` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | N |
| `MAP` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N |
| `ROW` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N |
| `STRUCTURED` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N |
| `MAP` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | !⁸ | N |
| `ROW` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | !⁸ | N |
| `STRUCTURED` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | !⁸ | N |
| `RAW` | Y | ! | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | Y⁴ | N | N |
| `VARIANT` | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N |
| `BITMAP` | Y | Y⁷ | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N |
Expand All @@ -1971,6 +2006,7 @@ COALESCE(TRY_CAST('non-number' AS INT), 0) --- 结果返回数字 0 的 INT 格
5. 支持转换，当且仅当用使用 `INTERVAL` 做“月”到“年”的转换。
6. 支持转换，当且仅当用使用 `INTERVAL` 做“天”到“时间”的转换。
7. 仅支持转换到无界的 `VARBINARY`（`BYTES`），因为裁剪或填充会破坏序列化的位图数据。
8. Supported iff every leaf type casts to `VARIANT` and every `MAP` key is a character string. Always fallible for `ARRAY` and `MAP`. Fallible for `ROW` and `STRUCTURED` iff a field is fallible, a field is a `VARIANT`, or the declared sizes of the fields can exceed the 16 MiB of a `VARIANT`.

请注意：无论是 `CAST` 还是 `TRY_CAST`，当输入为 `NULL` ，输出也为 `NULL`。

Expand Down
4 changes: 2 additions & 2 deletions docs/content/docs/sql/reference/data-types.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1740,8 +1740,8 @@ leaf is stored by the rules above, so the cast is supported only when every leaf
- A `ROW` or `STRUCTURED` value is keyed by its field names.
- A `MAP` needs a character string key, which becomes the object key. A `NULL` key fails the cast.
If a key appears twice, the last value is kept.
- A variant object sorts its keys, so neither the field order of a `ROW` nor the entry order of a
`MAP` is kept.
- A variant object sorts its keys by their UTF-8 bytes, so neither the field order of a `ROW` nor
the entry order of a `MAP` is kept.
- A `NULL` element, field, or map value becomes a variant null, so an array keeps its length and an
object keeps its keys. A cast element by element differs here: `ARRAY<INT>` to `ARRAY<VARIANT>`
keeps a SQL `NULL`, while `ARRAY<INT>` to `VARIANT` turns it into a variant null.
Expand Down
81 changes: 52 additions & 29 deletions flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,10 +37,14 @@
import static org.apache.flink.types.variant.BinaryVariantUtil.VERSION;
import static org.apache.flink.types.variant.BinaryVariantUtil.VERSION_MASK;
import static org.apache.flink.types.variant.BinaryVariantUtil.checkIndex;
import static org.apache.flink.types.variant.BinaryVariantUtil.compareMetadataKey;
import static org.apache.flink.types.variant.BinaryVariantUtil.encodeKey;
import static org.apache.flink.types.variant.BinaryVariantUtil.getMetadataKey;
import static org.apache.flink.types.variant.BinaryVariantUtil.handleArray;
import static org.apache.flink.types.variant.BinaryVariantUtil.handleObject;
import static org.apache.flink.types.variant.BinaryVariantUtil.hasCodePointFromE000;
import static org.apache.flink.types.variant.BinaryVariantUtil.malformedVariant;
import static org.apache.flink.types.variant.BinaryVariantUtil.metadataKeyEquals;
import static org.apache.flink.types.variant.BinaryVariantUtil.microsToInstant;
import static org.apache.flink.types.variant.BinaryVariantUtil.nanosToInstant;
import static org.apache.flink.types.variant.BinaryVariantUtil.readUnsigned;
Expand Down Expand Up @@ -337,48 +341,67 @@ private void checkType(Type expected, Type actual) {
// Find the field value whose key is equal to `key`. Return null if the key is not found.
// It is only legal to call it when `getType()` is `Type.OBJECT`.
private BinaryVariant getFieldByKey(String key) {
byte[] keyBytes = encodeKey(key);
return handleObject(
value,
pos,
(size, idSize, offsetSize, idStart, offsetStart, dataStart) -> {
int index;
// Use linear search for a short list. Switch to binary search when the length
// reaches `BINARY_SEARCH_THRESHOLD`.
if (size < BINARY_SEARCH_THRESHOLD) {
for (int i = 0; i < size; ++i) {
int id = readUnsigned(value, idStart + idSize * i, idSize);
if (key.equals(getMetadataKey(metadata, id))) {
int offset =
readUnsigned(
value, offsetStart + offsetSize * i, offsetSize);
return new BinaryVariant(value, metadata, dataStart + offset);
}
}
index = findFieldLinear(keyBytes, size, idSize, idStart);
} else {
int low = 0;
int high = size - 1;
while (low <= high) {
// Use unsigned right shift to compute the middle of `low` and `high`.
// This is not only a performance optimization, because it can properly
// handle the case where `low + high` overflows int.
int mid = (low + high) >>> 1;
int id = readUnsigned(value, idStart + idSize * mid, idSize);
int cmp = getMetadataKey(metadata, id).compareTo(key);
if (cmp < 0) {
low = mid + 1;
} else if (cmp > 0) {
high = mid - 1;
} else {
int offset =
readUnsigned(
value, offsetStart + offsetSize * mid, offsetSize);
return new BinaryVariant(value, metadata, dataStart + offset);
}
index = findFieldBinary(keyBytes, size, idSize, idStart);
// Objects that earlier Flink versions wrote are sorted in UTF-16 order
// instead. Only a miss for a key with a code point from U+E000 up can be
// wrong there, see `hasCodePointFromE000`. Such a miss scans all fields,
// because an object does not record which order it uses.
if (index < 0 && hasCodePointFromE000(keyBytes)) {
index = findFieldLinear(keyBytes, size, idSize, idStart);
}
}
return null;
if (index < 0) {
return null;
}
int offset = readUnsigned(value, offsetStart + offsetSize * index, offsetSize);
return new BinaryVariant(value, metadata, dataStart + offset);
});
}

// Return the index of the field whose key has the UTF-8 bytes `key`, or -1 if there is none.
private int findFieldLinear(byte[] key, int size, int idSize, int idStart) {
for (int i = 0; i < size; ++i) {
int id = readUnsigned(value, idStart + idSize * i, idSize);
if (metadataKeyEquals(metadata, id, key)) {
return i;
}
}
return -1;
}

// Like `findFieldLinear`, for fields sorted by the unsigned UTF-8 bytes of their keys.
private int findFieldBinary(byte[] key, int size, int idSize, int idStart) {
int low = 0;
int high = size - 1;
while (low <= high) {
// Use unsigned right shift to compute the middle of `low` and `high`. This is not only
// a performance optimization, because it can properly handle the case where
// `low + high` overflows int.
int mid = (low + high) >>> 1;
int id = readUnsigned(value, idStart + idSize * mid, idSize);
int cmp = compareMetadataKey(metadata, id, key);
if (cmp < 0) {
low = mid + 1;
} else if (cmp > 0) {
high = mid - 1;
} else {
return mid;
}
}
return -1;
}

// Get the array element at the `index` slot. Return null if `index` is out of the bound of
// `[0, arraySize())`.
// It is only legal to call it when `getType()` is `Type.ARRAY`.
Expand Down
4 changes: 1 addition & 3 deletions flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,9 +186,7 @@ public VariantObjectBuilder(boolean allowDuplicateKeys) {

@Override
public VariantObjectBuilder add(String key, Variant value) {
int id = builder.addKey(key);
entries.add(
new BinaryVariantInternalBuilder.FieldEntry(key, id, builder.getWritePos()));
builder.addField(entries, 0, key);
builder.appendVariant((BinaryVariant) value);
return this;
}
Expand Down
Loading