raminqaf commented on code in PR #29636:
URL: https://github.com/apache/flink/pull/29636#discussion_r4388725241
##########
flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java:
##########
@@ -341,32 +347,68 @@ 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) {
- index = findFieldLinear(keyBytes, size, idSize,
idStart);
- } else {
- 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);
- }
- }
- if (index < 0) {
- return null;
- }
- int offset = readUnsigned(value, offsetStart + offsetSize
* index, offsetSize);
- return new BinaryVariant(value, metadata, dataStart +
offset);
- });
+ return getFieldByKey(encodeKey(key), null);
+ }
+
+ /**
+ * Returns the field whose key has the UTF-8 bytes {@code key}, or null if
there is none.
+ *
+ * <p>A {@code getter} remembers the index at which it last found its key.
That field is checked
+ * first, so a correct guess skips the search and a wrong one costs one
key comparison.
+ *
+ * @throws VariantTypeException if the variant is not an object
+ */
+ @Nullable
+ BinaryVariant getFieldByKey(byte[] key, @Nullable VariantFieldGetter
getter) {
+ // Decode the header here instead of in `handleObject` and its
capturing lambda, so that a
+ // lookup allocates nothing but the returned view.
+ int typeInfo = objectTypeInfo(value, pos);
+ int sizeBytes = objectSizeBytes(typeInfo);
+ int size = readUnsigned(value, pos + 1, sizeBytes);
+ int idSize = objectIdSize(typeInfo);
+ int idStart = pos + 1 + sizeBytes;
+ // Read the remembered index once, so that a getter shared between
threads only ever
+ // returns a field whose key it checked.
+ int index = getter == null ? 0 : getter.lastIndex;
+ if (getter == null || !hasKeyAt(key, index, size, idSize, idStart)) {
+ index = findField(key, size, idSize, idStart);
+ if (index < 0) {
+ return null;
+ }
+ if (getter != null) {
+ getter.lastIndex = index;
+ }
+ }
+ int offsetSize = objectOffsetSize(typeInfo);
+ int offsetStart = idStart + size * idSize;
+ int dataStart = offsetStart + (size + 1) * offsetSize;
+ int offset = readUnsigned(value, offsetStart + offsetSize * index,
offsetSize);
+ return new BinaryVariant(value, metadata, dataStart + offset);
+ }
+
+ private boolean hasKeyAt(byte[] key, int index, int size, int idSize, int
idStart) {
+ return index < size
+ && metadataKeyEquals(
+ metadata, readUnsigned(value, idStart + idSize *
index, idSize), key);
+ }
+
+ /**
+ * Like {@code findFieldLinear}, but uses binary search from {@code
BINARY_SEARCH_THRESHOLD}
+ * fields on.
+ */
+ private int findField(byte[] key, int size, int idSize, int idStart) {
+ if (size < BINARY_SEARCH_THRESHOLD) {
+ return findFieldLinear(key, size, idSize, idStart);
+ }
+ int index = findFieldBinary(key, 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(key)) {
+ index = findFieldLinear(key, size, idSize, idStart);
+ }
+ return index;
}
Review Comment:
Only refactored and moved into a new method
--
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]
