snuyanzin commented on code in PR #29587:
URL: https://github.com/apache/flink/pull/29587#discussion_r4381805220
##########
flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java:
##########
@@ -621,53 +623,122 @@ public void appendVariant(BinaryVariant v) {
appendVariantImpl(v.rawValue(), v.getMetadata(), v.getPos());
}
+ // Uses an explicit stack rather than recursion, so deeply nested input
cannot overflow.
private void appendVariantImpl(byte[] value, byte[] metadata, int pos) {
checkIndex(pos, value.length);
- int basicType = value[pos] & BASIC_TYPE_MASK;
- switch (basicType) {
- case OBJECT:
- handleObject(
- value,
- pos,
- (size, idSize, offsetSize, idStart, offsetStart,
dataStart) -> {
- ArrayList<FieldEntry> fields = new
ArrayList<>(size);
- int start = writePos;
- for (int i = 0; i < size; ++i) {
- int id = readUnsigned(value, idStart + idSize
* i, idSize);
- int offset =
- readUnsigned(
- value, offsetStart +
offsetSize * i, offsetSize);
- int elementPos = dataStart + offset;
- String key = getMetadataKey(metadata, id);
- addField(fields, start, key);
- appendVariantImpl(value, metadata, elementPos);
- }
- finishWritingObject(start, fields);
- return null;
- });
- break;
- case ARRAY:
- handleArray(
- value,
- pos,
- (size, offsetSize, offsetStart, dataStart) -> {
- ArrayList<Integer> offsets = new ArrayList<>(size);
- int start = writePos;
- for (int i = 0; i < size; ++i) {
- int offset =
- readUnsigned(
- value, offsetStart +
offsetSize * i, offsetSize);
- int elementPos = dataStart + offset;
- offsets.add(writePos - start);
- appendVariantImpl(value, metadata, elementPos);
- }
- finishWritingArray(start, offsets);
- return null;
- });
- break;
- default:
- shallowAppendVariantImpl(value, pos);
- break;
+ int rootType = value[pos] & BASIC_TYPE_MASK;
+ if (rootType != OBJECT && rootType != ARRAY) {
+ // Scalars are the common case on per-record paths, so skip
allocating the stack.
+ shallowAppendVariantImpl(value, pos);
+ return;
+ }
+ final Deque<AppendFrame> stack = new ArrayDeque<>();
+ int next = pos;
+ while (true) {
+ if (next >= 0) {
+ checkIndex(next, value.length);
+ int basicType = value[next] & BASIC_TYPE_MASK;
+ switch (basicType) {
+ case OBJECT:
+ stack.push(
+ handleObject(
+ value,
+ next,
+ (size,
+ idSize,
+ offsetSize,
+ idStart,
+ offsetStart,
+ dataStart) ->
+ new AppendFrame(
+ size,
+ idSize,
+ idStart,
+ offsetSize,
+ offsetStart,
+ dataStart,
+ writePos,
+ true)));
+ break;
+ case ARRAY:
+ stack.push(
+ handleArray(
+ value,
+ next,
+ (size, offsetSize, offsetStart,
dataStart) ->
+ new AppendFrame(
+ size,
+ 0,
+ 0,
+ offsetSize,
+ offsetStart,
+ dataStart,
+ writePos,
+ false)));
+ break;
+ default:
+ shallowAppendVariantImpl(value, next);
+ break;
+ }
+ }
+ final AppendFrame frame = stack.peek();
+ if (frame == null) {
+ return;
+ }
+ if (frame.index == frame.size) {
+ stack.pop();
+ if (frame.fields != null) {
+ finishWritingObject(frame.start, frame.fields);
+ } else {
+ finishWritingArray(frame.start, frame.offsets);
+ }
+ next = -1;
+ continue;
+ }
+ int i = frame.index++;
+ int offset =
+ readUnsigned(value, frame.offsetStart + frame.offsetSize *
i, frame.offsetSize);
+ if (frame.fields != null) {
+ int id = readUnsigned(value, frame.idStart + frame.idSize * i,
frame.idSize);
+ addField(frame.fields, frame.start, getMetadataKey(metadata,
id));
+ } else {
+ frame.offsets.add(writePos - frame.start);
+ }
+ next = frame.dataStart + offset;
Review Comment:
added
--
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]
