Changes

Author: Tim-Brooks

server/src/main/java/org/elasticsearch/ingest/ESONBytes.java

@@ -0,0 +1,12 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.ingest;
+
+public class ESONBytes {}

server/src/main/java/org/elasticsearch/ingest/ESONXContentParser.java

@@ -112,7 +112,9 @@ public void allowDuplicateKeys(boolean allowDuplicateKeys) {
 
     @Override
     public Token nextToken() throws IOException {
-        if (closed) return null;
+        if (closed) {
+            return null;
+        }
 
         // Clear value state
         currentValue = null;
@@ -132,7 +134,8 @@ public Token nextToken() throws IOException {
 
             case OBJECT_FIELD_NAME:
                 if (currentContainer.fieldsRemaining == 0) {
-                    Token endToken = currentContainer.type == ContainerContext.Type.OBJECT ? Token.END_OBJECT : Token.END_ARRAY;
+                    assert currentContainer.type == ContainerContext.Type.OBJECT;
+                    Token endToken = Token.END_OBJECT;
                     containerStack.pop();
                     currentContainer = containerStack.isEmpty() ? null : containerStack.peek();
                     updateStateAfterEnd();
@@ -143,8 +146,18 @@ public Token nextToken() throws IOException {
                     return currentToken = Token.FIELD_NAME;
                 }
             case OBJECT_VALUE:
-            case ARRAY_VALUE:
                 return emitValue(keyArray.get(currentIndex++));
+            case ARRAY_VALUE:
+                if (currentContainer.fieldsRemaining == 0) {
+                    assert currentContainer.type == ContainerContext.Type.ARRAY;
+                    Token endToken = Token.END_ARRAY;
+                    containerStack.pop();
+                    currentContainer = containerStack.isEmpty() ? null : containerStack.peek();
+                    updateStateAfterEnd();
+                    return currentToken = endToken;
+                } else {
+                    return emitValue(keyArray.get(currentIndex++));
+                }
             case DONE:
                 return null;
         }
@@ -168,19 +181,21 @@ private Token emitValue(ESONSource.KeyEntry entry) {
             case ESONSource.KeyEntry.TYPE_OBJECT -> {
                 containerStack.push(new ContainerContext(ContainerContext.Type.OBJECT, ((ESONSource.ObjectEntry) entry).fieldCount));
                 currentToken = Token.START_OBJECT;
+                state = State.OBJECT_FIELD_NAME;
             }
             case ESONSource.KeyEntry.TYPE_ARRAY -> {
                 containerStack.push(new ContainerContext(ContainerContext.Type.ARRAY, ((ESONSource.ArrayEntry) entry).elementCount));
                 currentToken = Token.START_ARRAY;
+                state = State.ARRAY_VALUE;
             }
             case ESONSource.KeyEntry.TYPE_FIELD -> {
                 currentType = ((ESONSource.FieldEntry) entry).type;
                 currentToken = determineValueToken(currentType);
+                state = State.OBJECT_FIELD_NAME;
             }
             default -> throw new IllegalStateException("Unknown key entry type: " + entry.type());
         }
 
-        currentIndex++;
         if (currentContainer != null) {
             currentContainer.fieldsRemaining--;
         }