content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.arrow.vector.ipc;
  
  import static com.fasterxml.jackson.core.JsonToken.END_ARRAY;
  import static com.fasterxml.jackson.core.JsonToken.END_OBJECT;
  import static com.fasterxml.jackson.core.JsonToken.START_ARRAY;
  import static com.fasterxml.jackson.core.JsonToken.START_OBJECT;
  import static org.apache.arrow.vector.BitVectorHelper.getValidityBufferSizeFromCount;
  import static org.apache.arrow.vector.BufferLayout.BufferType.DATA;
  import static org.apache.arrow.vector.BufferLayout.BufferType.OFFSET;
  import static org.apache.arrow.vector.BufferLayout.BufferType.SIZE;
  import static org.apache.arrow.vector.BufferLayout.BufferType.TYPE;
  import static org.apache.arrow.vector.BufferLayout.BufferType.VALIDITY;
  import static org.apache.arrow.vector.BufferLayout.BufferType.VARIADIC_DATA_BUFFERS;
  import static org.apache.arrow.vector.BufferLayout.BufferType.VIEWS;
  
  import com.fasterxml.jackson.core.JsonParseException;
  import com.fasterxml.jackson.core.JsonParser;
  import com.fasterxml.jackson.core.JsonParser.Feature;
  import com.fasterxml.jackson.core.JsonToken;
  import com.fasterxml.jackson.databind.MapperFeature;
  import com.fasterxml.jackson.databind.MappingJsonFactory;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import java.io.File;
  import java.io.IOException;
  import java.math.BigDecimal;
  import java.math.BigInteger;
  import java.nio.ByteBuffer;
  import java.nio.ByteOrder;
  import java.nio.charset.StandardCharsets;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import org.apache.arrow.memory.ArrowBuf;
  import org.apache.arrow.memory.BufferAllocator;
  import org.apache.arrow.util.Preconditions;
  import org.apache.arrow.vector.BaseVariableWidthViewVector;
  import org.apache.arrow.vector.BigIntVector;
  import org.apache.arrow.vector.BitVectorHelper;
  import org.apache.arrow.vector.BufferLayout.BufferType;
  import org.apache.arrow.vector.Decimal256Vector;
  import org.apache.arrow.vector.DecimalVector;
  import org.apache.arrow.vector.FieldVector;
  import org.apache.arrow.vector.Float4Vector;
  import org.apache.arrow.vector.Float8Vector;
  import org.apache.arrow.vector.IntVector;
  import org.apache.arrow.vector.IntervalDayVector;
  import org.apache.arrow.vector.IntervalMonthDayNanoVector;
  import org.apache.arrow.vector.SmallIntVector;
  import org.apache.arrow.vector.TinyIntVector;
  import org.apache.arrow.vector.TypeLayout;
  import org.apache.arrow.vector.VectorSchemaRoot;
  import org.apache.arrow.vector.dictionary.Dictionary;
  import org.apache.arrow.vector.dictionary.DictionaryProvider;
  import org.apache.arrow.vector.ipc.message.ArrowFieldNode;
  import org.apache.arrow.vector.types.Types.MinorType;
  import org.apache.arrow.vector.types.pojo.ArrowType;
  import org.apache.arrow.vector.types.pojo.ArrowType.LargeListView;
  import org.apache.arrow.vector.types.pojo.ArrowType.ListView;
  import org.apache.arrow.vector.types.pojo.ArrowType.Union;
  import org.apache.arrow.vector.types.pojo.Field;
  import org.apache.arrow.vector.types.pojo.Schema;
  import org.apache.arrow.vector.util.DecimalUtility;
  import org.apache.arrow.vector.util.DictionaryUtility;
  import org.apache.commons.codec.DecoderException;
  import org.apache.commons.codec.binary.Hex;
  
  /**
   * A reader for JSON files that translates them into vectors. This reader is used for integration
   * tests.
   *
   * <p>This class uses a streaming parser API, method naming tends to reflect this implementation
   * detail.
   */
  public class JsonFileReader implements AutoCloseable, DictionaryProvider {
    private final JsonParser parser;
    private final BufferAllocator allocator;
    private Schema schema;
    private Map<Long, Dictionary> dictionaries;
    private Boolean started = false;
  
    /**
     * Constructs a new instance.
     *
     * @param inputFile The file to read.
     * @param allocator The allocator to use for allocating buffers.
     */
    public JsonFileReader(File inputFile, BufferAllocator allocator)
        throws JsonParseException, IOException {
      super();
      this.allocator = allocator;
      MappingJsonFactory jsonFactory =
          new MappingJsonFactory(
              new ObjectMapper()
                  // ignore case for enums
                  .configure(MapperFeature.ACCEPT_CASE_INSENSITIVE_ENUMS, true));
      this.parser = jsonFactory.createParser(inputFile);
      // Allow reading NaN for floating point values
      this.parser.configure(Feature.ALLOW_NON_NUMERIC_NUMBERS, true);
    }
  
    @Override
    public Dictionary lookup(long id) {
      if (!started) {
        throw new IllegalStateException("Unable to lookup until after read() has started");
      }
  
      return dictionaries.get(id);
    }
  
    @Override
    public Set<Long> getDictionaryIds() {
      return dictionaries.keySet();
    }
  
    /** Reads the beginning (schema section) of the json file and returns it. */
    public Schema start() throws JsonParseException, IOException {
      readToken(START_OBJECT);
      {
        Schema originalSchema = readNextField("schema", Schema.class);
        List<Field> fields = new ArrayList<>();
        dictionaries = new HashMap<>();
  
        // Convert fields with dictionaries to have the index type
        for (Field field : originalSchema.getFields()) {
          fields.add(DictionaryUtility.toMemoryFormat(field, allocator, dictionaries));
        }
        this.schema = new Schema(fields, originalSchema.getCustomMetadata());
  
        if (!dictionaries.isEmpty()) {
          nextFieldIs("dictionaries");
          readDictionaryBatches();
        }
  
        nextFieldIs("batches");
        readToken(START_ARRAY);
        started = true;
        return this.schema;
      }
    }
  
    private void readDictionaryBatches() throws JsonParseException, IOException {
      readToken(START_ARRAY);
      JsonToken token = parser.nextToken();
      boolean haveDictionaryBatch = token == START_OBJECT;
      while (haveDictionaryBatch) {
  
        // Lookup what dictionary for the batch about to be read
        long id = readNextField("id", Long.class);
        Dictionary dict = dictionaries.get(id);
        if (dict == null) {
          throw new IllegalArgumentException(
              "Dictionary with id: " + id + " missing encoding from schema Field");
        }
  
        // Read the dictionary record batch
        nextFieldIs("data");
        FieldVector vector = dict.getVector();
        List<Field> fields = Collections.singletonList(vector.getField());
        List<FieldVector> vectors = Collections.singletonList(vector);
        VectorSchemaRoot root = new VectorSchemaRoot(fields, vectors, vector.getValueCount());
        read(root);
  
        readToken(END_OBJECT);
        token = parser.nextToken();
        haveDictionaryBatch = token == START_OBJECT;
      }
  
      if (token != END_ARRAY) {
        throw new IllegalArgumentException(
            "Invalid token: " + token + " expected end of array at " + parser.getTokenLocation());
      }
    }
  
    /** Reads the next record batch from the file into <code>root</code>. */
    public boolean read(VectorSchemaRoot root) throws IOException {
      JsonToken t = parser.nextToken();
      if (t == START_OBJECT) {
        {
          int count = readNextField("count", Integer.class);
          nextFieldIs("columns");
          readToken(START_ARRAY);
          {
            for (Field field : root.getSchema().getFields()) {
              FieldVector vector = root.getVector(field);
              readFromJsonIntoVector(field, vector);
            }
          }
          readToken(END_ARRAY);
          root.setRowCount(count);
        }
        readToken(END_OBJECT);
        return true;
      } else if (t == END_ARRAY) {
        root.setRowCount(0);
        return false;
      } else {
        throw new IllegalArgumentException("Invalid token: " + t);
      }
    }
  
    /** Returns the next record batch from the file. */
    public VectorSchemaRoot read() throws IOException {
      JsonToken t = parser.nextToken();
      if (t == START_OBJECT) {
        VectorSchemaRoot recordBatch = VectorSchemaRoot.create(schema, allocator);
        {
          int count = readNextField("count", Integer.class);
          recordBatch.setRowCount(count);
          nextFieldIs("columns");
          readToken(START_ARRAY);
          {
            for (Field field : schema.getFields()) {
              FieldVector vector = recordBatch.getVector(field);
              readFromJsonIntoVector(field, vector);
            }
          }
          readToken(END_ARRAY);
        }
        readToken(END_OBJECT);
        return recordBatch;
      } else if (t == END_ARRAY) {
        return null;
      } else {
        throw new IllegalArgumentException("Invalid token: " + t);
      }
    }
  
    /**
     * Skips a number of record batches in the file.
     *
     * @param numBatches the number of batches to skip
     * @return the actual number of skipped batches.
     */
    // This is currently called using JPype by the integration tests.
    public int skip(int numBatches) throws IOException {
      for (int i = 0; i < numBatches; ++i) {
        JsonToken t = parser.nextToken();
        if (t == START_OBJECT) {
          parser.skipChildren();
          assert parser.getCurrentToken() == END_OBJECT;
        } else if (t == END_ARRAY) {
          return i;
        } else {
          throw new IllegalArgumentException("Invalid token: " + t);
        }
      }
      return numBatches;
    }
  
    private abstract class BufferReader {
      protected abstract ArrowBuf read(BufferAllocator allocator, int count) throws IOException;
  
      ArrowBuf readBuffer(BufferAllocator allocator, int count) throws IOException {
        readToken(START_ARRAY);
        ArrowBuf buf = read(allocator, count);
        readToken(END_ARRAY);
        return buf;
      }
    }
  
    /**
     * Read all the variadic data buffers from the parser.
     *
     * @param allocator BufferAllocator
     * @param variadicBuffersCount Number of variadic buffers
     * @return List of ArrowBuf
     * @throws IOException throws IOException in a failure
     */
    List<ArrowBuf> readVariadicBuffers(BufferAllocator allocator, int variadicBuffersCount)
        throws IOException {
      readToken(START_ARRAY);
      ArrayList<ArrowBuf> dataBuffers = new ArrayList<>(variadicBuffersCount);
      for (int i = 0; i < variadicBuffersCount; i++) {
        parser.nextToken();
        final byte[] value;
  
        String variadicStr = parser.readValueAs(String.class);
        if (variadicStr == null) {
          value = new byte[0];
        } else {
          value = decodeHexSafe(variadicStr);
        }
  
        ArrowBuf buf = allocator.buffer(value.length);
        buf.writeBytes(value);
        dataBuffers.add(buf);
      }
      readToken(END_ARRAY);
      return dataBuffers;
    }
  
    private ArrowBuf readViewBuffers(
        BufferAllocator allocator, int count, List<Integer> variadicBufferIndices, MinorType type)
        throws IOException {
      readToken(START_ARRAY);
      ArrayList<byte[]> values = new ArrayList<>(count);
      long bufferSize = 0L;
      for (int i = 0; i < count; i++) {
        readToken(START_OBJECT);
        final int length = readNextField("SIZE", Integer.class);
        byte[] value;
        if (length > BaseVariableWidthViewVector.INLINE_SIZE) {
          // PREFIX_HEX
          final byte[] prefix = decodeHexSafe(readNextField("PREFIX_HEX", String.class));
          // BUFFER_INDEX
          final int bufferIndex = readNextField("BUFFER_INDEX", Integer.class);
          if (variadicBufferIndices.isEmpty()) {
            variadicBufferIndices.add(bufferIndex);
          } else {
            int lastBufferIndex = variadicBufferIndices.get(variadicBufferIndices.size() - 1);
            if (lastBufferIndex != bufferIndex) {
              variadicBufferIndices.add(bufferIndex);
            }
          }
  
          // OFFSET
          final int offset = readNextField("OFFSET", Integer.class);
          ByteBuffer buffer =
              ByteBuffer.allocate(BaseVariableWidthViewVector.ELEMENT_SIZE)
                  .order(ByteOrder.LITTLE_ENDIAN); // Allocate a ByteBuffer of size 16 bytes
          buffer.putInt(length); // Write 'length' to bytes 0-3
          buffer.put(prefix); // Write 'prefix' to bytes 4-7
          buffer.putInt(bufferIndex); // Write 'bufferIndex' to bytes 8-11
          buffer.putInt(offset); // Write 'offset' to bytes 12-15
          value = buffer.array(); // Convert the ByteBuffer to a byte array
        } else {
          // in-line
          ByteBuffer buffer =
              ByteBuffer.allocate(BaseVariableWidthViewVector.ELEMENT_SIZE)
                  .order(ByteOrder.LITTLE_ENDIAN); // Allocate a ByteBuffer of size 16 bytes
          buffer.putInt(length); // Write 'length' to bytes 0-3
          // INLINE
          if (type == MinorType.VIEWVARCHAR) {
            buffer.put(readNextField("INLINED", String.class).getBytes(StandardCharsets.UTF_8));
          } else {
            String inlined = readNextField("INLINED", String.class);
            if (inlined == null) {
              buffer.put(new byte[length]);
            } else {
              buffer.put(decodeHexSafe(inlined));
            }
          }
          value = buffer.array(); // Convert the ByteBuffer to a byte array
        }
        values.add(value);
        bufferSize += value.length;
        readToken(END_OBJECT);
      }
  
      ArrowBuf buf = allocator.buffer(bufferSize);
  
      for (byte[] value : values) {
        buf.writeBytes(value);
      }
      readToken(END_ARRAY);
      return buf;
    }
  
    private class BufferHelper {
  
      BufferReader BIT =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final int bufferSize = getValidityBufferSizeFromCount(count);
              ArrowBuf buf = allocator.buffer(bufferSize);
  
              // C++ integration test fails without this.
              buf.setZero(0, bufferSize);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                BitVectorHelper.setValidityBit(buf, i, parser.readValueAs(Boolean.class) ? 1 : 0);
              }
  
              buf.writerIndex(bufferSize);
              return buf;
            }
          };
  
      BufferReader DAY_MILLIS =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * IntervalDayVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                readToken(START_OBJECT);
                buf.writeInt(readNextField("days", Integer.class));
                buf.writeInt(readNextField("milliseconds", Integer.class));
                readToken(END_OBJECT);
              }
  
              return buf;
            }
          };
  
      BufferReader MONTH_DAY_NANOS =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * IntervalMonthDayNanoVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                readToken(START_OBJECT);
                buf.writeInt(readNextField("months", Integer.class));
                buf.writeInt(readNextField("days", Integer.class));
                buf.writeLong(readNextField("nanoseconds", Long.class));
                readToken(END_OBJECT);
              }
  
              return buf;
            }
          };
  
      BufferReader INT1 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * TinyIntVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                buf.writeByte(parser.getByteValue());
              }
  
              return buf;
            }
          };
  
      BufferReader INT2 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * SmallIntVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                buf.writeShort(parser.getShortValue());
              }
  
              return buf;
            }
          };
  
      BufferReader INT4 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * IntVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                buf.writeInt(parser.getIntValue());
              }
  
              return buf;
            }
          };
  
      BufferReader INT8 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * BigIntVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                String value = parser.getValueAsString();
                buf.writeLong(Long.valueOf(value));
              }
  
              return buf;
            }
          };
  
      BufferReader UINT1 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * TinyIntVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                buf.writeByte(parser.getShortValue() & 0xFF);
              }
  
              return buf;
            }
          };
  
      BufferReader UINT2 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * SmallIntVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                buf.writeShort(parser.getIntValue() & 0xFFFF);
              }
  
              return buf;
            }
          };
  
      BufferReader UINT4 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * IntVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                buf.writeInt((int) parser.getLongValue());
              }
  
              return buf;
            }
          };
  
      BufferReader UINT8 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * BigIntVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                BigInteger value = new BigInteger(parser.getValueAsString());
                buf.writeLong(value.longValue());
              }
  
              return buf;
            }
          };
  
      BufferReader FLOAT4 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * Float4Vector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                buf.writeFloat(parser.getFloatValue());
              }
  
              return buf;
            }
          };
  
      BufferReader FLOAT8 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * Float8Vector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                buf.writeDouble(parser.getDoubleValue());
              }
  
              return buf;
            }
          };
  
      BufferReader DECIMAL =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * DecimalVector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                BigDecimal decimalValue = new BigDecimal(parser.readValueAs(String.class));
                DecimalUtility.writeBigDecimalToArrowBuf(
                    decimalValue, buf, i, DecimalVector.TYPE_WIDTH);
              }
  
              buf.writerIndex(size);
              return buf;
            }
          };
  
      BufferReader DECIMAL256 =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              final long size = (long) count * Decimal256Vector.TYPE_WIDTH;
              ArrowBuf buf = allocator.buffer(size);
  
              for (int i = 0; i < count; i++) {
                parser.nextToken();
                BigDecimal decimalValue = new BigDecimal(parser.readValueAs(String.class));
                DecimalUtility.writeBigDecimalToArrowBuf(
                    decimalValue, buf, i, Decimal256Vector.TYPE_WIDTH);
              }
  
              buf.writerIndex(size);
              return buf;
            }
          };
  
      ArrowBuf readBinaryValues(BufferAllocator allocator, int count) throws IOException {
        ArrayList<byte[]> values = new ArrayList<>(count);
        long bufferSize = 0L;
        for (int i = 0; i < count; i++) {
          parser.nextToken();
          final byte[] value = decodeHexSafe(parser.readValueAs(String.class));
          values.add(value);
          bufferSize += value.length;
        }
  
        ArrowBuf buf = allocator.buffer(bufferSize);
  
        for (byte[] value : values) {
          buf.writeBytes(value);
        }
  
        return buf;
      }
  
      ArrowBuf readStringValues(BufferAllocator allocator, int count) throws IOException {
        ArrayList<byte[]> values = new ArrayList<>(count);
        long bufferSize = 0L;
        for (int i = 0; i < count; i++) {
          parser.nextToken();
          final byte[] value = parser.getValueAsString().getBytes(StandardCharsets.UTF_8);
          values.add(value);
          bufferSize += value.length;
        }
  
        ArrowBuf buf = allocator.buffer(bufferSize);
  
        for (byte[] value : values) {
          buf.writeBytes(value);
        }
  
        return buf;
      }
  
      BufferReader FIXEDSIZEBINARY =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              return readBinaryValues(allocator, count);
            }
          };
  
      BufferReader VARCHAR =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              return readStringValues(allocator, count);
            }
          };
  
      BufferReader LARGEVARCHAR =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              return readStringValues(allocator, count);
            }
          };
  
      BufferReader VARBINARY =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              return readBinaryValues(allocator, count);
            }
          };
  
      BufferReader LARGEVARBINARY =
          new BufferReader() {
            @Override
            protected ArrowBuf read(BufferAllocator allocator, int count) throws IOException {
              return readBinaryValues(allocator, count);
            }
          };
    }
  
    private List<ArrowBuf> readIntoBuffer(
        BufferAllocator allocator,
        BufferType bufferType,
        MinorType type,
        int count,
        List<Integer> variadicBufferIndices)
        throws IOException {
      ArrowBuf buf;
  
      BufferHelper helper = new BufferHelper();
      BufferReader reader;
  
      if (bufferType.equals(VALIDITY)) {
        reader = helper.BIT;
      } else if (bufferType.equals(OFFSET) || bufferType.equals(SIZE)) {
        if (type == MinorType.LARGELIST
            || type == MinorType.LARGEVARCHAR
            || type == MinorType.LARGEVARBINARY
            || type == MinorType.LARGELISTVIEW) {
          reader = helper.INT8;
        } else {
          reader = helper.INT4;
        }
      } else if (bufferType.equals(TYPE)) {
        reader = helper.INT1;
      } else if (bufferType.equals(DATA)) {
        switch (type) {
          case BIT:
            reader = helper.BIT;
            break;
          case TINYINT:
            reader = helper.INT1;
            break;
          case SMALLINT:
            reader = helper.INT2;
            break;
          case INT:
            reader = helper.INT4;
            break;
          case BIGINT:
            reader = helper.INT8;
            break;
          case UINT1:
            reader = helper.UINT1;
            break;
          case UINT2:
            reader = helper.UINT2;
            break;
          case UINT4:
            reader = helper.UINT4;
            break;
          case UINT8:
            reader = helper.UINT8;
            break;
          case FLOAT4:
            reader = helper.FLOAT4;
            break;
          case FLOAT8:
            reader = helper.FLOAT8;
            break;
          case DECIMAL:
            reader = helper.DECIMAL;
            break;
          case DECIMAL256:
            reader = helper.DECIMAL256;
            break;
          case FIXEDSIZEBINARY:
            reader = helper.FIXEDSIZEBINARY;
            break;
          case VARCHAR:
            reader = helper.VARCHAR;
            break;
          case LARGEVARCHAR:
            reader = helper.LARGEVARCHAR;
            break;
          case VARBINARY:
            reader = helper.VARBINARY;
            break;
          case LARGEVARBINARY:
            reader = helper.LARGEVARBINARY;
            break;
          case DATEDAY:
            reader = helper.INT4;
            break;
          case DATEMILLI:
            reader = helper.INT8;
            break;
          case TIMESEC:
          case TIMEMILLI:
            reader = helper.INT4;
            break;
          case TIMEMICRO:
          case TIMENANO:
            reader = helper.INT8;
            break;
          case TIMESTAMPNANO:
          case TIMESTAMPMICRO:
          case TIMESTAMPMILLI:
          case TIMESTAMPSEC:
          case TIMESTAMPNANOTZ:
          case TIMESTAMPMICROTZ:
          case TIMESTAMPMILLITZ:
          case TIMESTAMPSECTZ:
            reader = helper.INT8;
            break;
          case INTERVALYEAR:
            reader = helper.INT4;
            break;
          case INTERVALDAY:
            reader = helper.DAY_MILLIS;
            break;
          case INTERVALMONTHDAYNANO:
            reader = helper.MONTH_DAY_NANOS;
            break;
          case DURATION:
            reader = helper.INT8;
            break;
          default:
            throw new UnsupportedOperationException("Cannot read array of type " + type);
        }
      } else if (bufferType.equals(VIEWS)) {
        return Collections.singletonList(
            readViewBuffers(allocator, count, variadicBufferIndices, type));
      } else if (bufferType.equals(VARIADIC_DATA_BUFFERS)) {
        return readVariadicBuffers(allocator, variadicBufferIndices.size());
      } else {
        throw new InvalidArrowFileException("Unrecognized buffer type " + bufferType);
      }
  
      buf = reader.readBuffer(allocator, count);
      Preconditions.checkNotNull(buf);
      return Collections.singletonList(buf);
    }
  
    private void readFromJsonIntoVector(Field field, FieldVector vector) throws IOException {
      ArrowType type = field.getType();
      TypeLayout typeLayout = TypeLayout.getTypeLayout(type);
      List<BufferType> vectorTypes = typeLayout.getBufferTypes();
      List<ArrowBuf> vectorBuffers = new ArrayList<>(vectorTypes.size());
      List<Integer> variadicBufferIndices = new ArrayList<>();
  
      if (!typeLayout.isFixedBufferCount()) {
        vectorTypes.add(VARIADIC_DATA_BUFFERS);
      }
      /*
       * The order of inner buffers is:
       * Fixed width vector:
       *    -- validity buffer
       *    -- data buffer
       * Variable width vector:
       *    -- validity buffer
       *    -- offset buffer
       *    -- data buffer
       *
       * This is similar to what getFieldInnerVectors() used to give but now that we don't have
       * inner vectors anymore, we will work directly at the buffer level -- populate buffers
       * locally as we read from Json parser and do loadFieldBuffers on the vector followed by
       * releasing the local buffers.
       */
      readToken(START_OBJECT);
      {
        // If currently reading dictionaries, field name is not important so don't check
        String name = readNextField("name", String.class);
        if (started && !Objects.equals(field.getName(), name)) {
          throw new IllegalArgumentException(
              "Expected field " + field.getName() + " but got " + name);
        }
  
        /* Initialize the vector with required capacity but don't allocateNew since we would
         * be doing loadFieldBuffers.
         */
        int valueCount = readNextField("count", Integer.class);
  
        vector.setInitialCapacity(valueCount);
  
        for (int v = 0; v < vectorTypes.size(); v++) {
          BufferType bufferType = vectorTypes.get(v);
          nextFieldIs(bufferType.getName());
          int innerBufferValueCount = valueCount;
          if (bufferType.equals(OFFSET)
              && !(type instanceof Union)
              && !(type instanceof ListView)
              && !(type instanceof LargeListView)) {
            /* offset buffer has 1 additional value capacity except for dense unions and ListView */
            innerBufferValueCount = valueCount + 1;
          }
  
          vectorBuffers.addAll(
              readIntoBuffer(
                  allocator,
                  bufferType,
                  vector.getMinorType(),
                  innerBufferValueCount,
                  variadicBufferIndices));
        }
  
        int nullCount;
        if (type instanceof ArrowType.RunEndEncoded || type instanceof Union) {
          nullCount = 0;
        } else if (type instanceof ArrowType.Null) {
          nullCount = valueCount;
        } else {
          nullCount = BitVectorHelper.getNullCount(vectorBuffers.get(0), valueCount);
        }
        final ArrowFieldNode fieldNode = new ArrowFieldNode(valueCount, nullCount);
        vector.loadFieldBuffers(fieldNode, vectorBuffers);
  
        /* read child vectors (if any) */
        List<Field> fields = field.getChildren();
        if (!fields.isEmpty()) {
          List<FieldVector> vectorChildren = vector.getChildrenFromFields();
          if (fields.size() != vectorChildren.size()) {
            throw new IllegalArgumentException(
                "fields and children are not the same size: "
                    + fields.size()
                    + " != "
                    + vectorChildren.size());
          }
          nextFieldIs("children");
          readToken(START_ARRAY);
          for (int i = 0; i < fields.size(); i++) {
            Field childField = fields.get(i);
            FieldVector childVector = vectorChildren.get(i);
            readFromJsonIntoVector(childField, childVector);
          }
          readToken(END_ARRAY);
        }
      }
  
      readToken(END_OBJECT);
  
      for (ArrowBuf buffer : vectorBuffers) {
        buffer.getReferenceManager().release();
      }
    }
  
    private byte[] decodeHexSafe(String hexString) throws IOException {
      try {
        return Hex.decodeHex(hexString.toCharArray());
      } catch (DecoderException e) {
        throw new IOException("Unable to decode hex string: " + hexString, e);
      }
    }
  
    @Override
    public void close() throws IOException {
      parser.close();
      if (dictionaries != null) {
        for (Dictionary dictionary : dictionaries.values()) {
          dictionary.getVector().close();
        }
      }
    }
  
    private <T> T readNextField(String expectedFieldName, Class<T> c)
        throws IOException, JsonParseException {
      nextFieldIs(expectedFieldName);
      parser.nextToken();
      return parser.readValueAs(c);
    }
  
    private void nextFieldIs(String expectedFieldName) throws IOException, JsonParseException {
      String name = parser.nextFieldName();
      if (name == null || !name.equals(expectedFieldName)) {
        throw new IllegalStateException("Expected " + expectedFieldName + " but got " + name);
      }
    }
  
    private void readToken(JsonToken expected) throws JsonParseException, IOException {
      JsonToken t = parser.nextToken();
      if (t != expected) {
        throw new IllegalStateException("Expected " + expected + " but got " + t);
      }
    }
  }
  