/*
 * 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.adapter.jdbc;

import static org.assertj.core.api.Assertions.assertThat;

import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.sql.Date;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Time;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
import org.apache.arrow.adapter.jdbc.binder.ColumnBinder;
import org.apache.arrow.memory.ArrowBuf;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.BaseLargeVariableWidthVector;
import org.apache.arrow.vector.BaseVariableWidthVector;
import org.apache.arrow.vector.BigIntVector;
import org.apache.arrow.vector.BitVector;
import org.apache.arrow.vector.DateDayVector;
import org.apache.arrow.vector.DateMilliVector;
import org.apache.arrow.vector.Decimal256Vector;
import org.apache.arrow.vector.DecimalVector;
import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.FixedSizeBinaryVector;
import org.apache.arrow.vector.Float4Vector;
import org.apache.arrow.vector.Float8Vector;
import org.apache.arrow.vector.IntVector;
import org.apache.arrow.vector.LargeVarBinaryVector;
import org.apache.arrow.vector.LargeVarCharVector;
import org.apache.arrow.vector.SmallIntVector;
import org.apache.arrow.vector.TimeMicroVector;
import org.apache.arrow.vector.TimeMilliVector;
import org.apache.arrow.vector.TimeNanoVector;
import org.apache.arrow.vector.TimeSecVector;
import org.apache.arrow.vector.TimeStampMicroTZVector;
import org.apache.arrow.vector.TimeStampMicroVector;
import org.apache.arrow.vector.TimeStampMilliTZVector;
import org.apache.arrow.vector.TimeStampMilliVector;
import org.apache.arrow.vector.TimeStampNanoTZVector;
import org.apache.arrow.vector.TimeStampNanoVector;
import org.apache.arrow.vector.TimeStampSecTZVector;
import org.apache.arrow.vector.TimeStampSecVector;
import org.apache.arrow.vector.TinyIntVector;
import org.apache.arrow.vector.VarBinaryVector;
import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.complex.ListVector;
import org.apache.arrow.vector.complex.MapVector;
import org.apache.arrow.vector.types.DateUnit;
import org.apache.arrow.vector.types.FloatingPointPrecision;
import org.apache.arrow.vector.types.TimeUnit;
import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Field;
import org.apache.arrow.vector.types.pojo.FieldType;
import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.arrow.vector.util.JsonStringHashMap;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

public class JdbcParameterBinderTest {
  private static final long MILLIS_PER_DAY = 86_400_000;
  BufferAllocator allocator;

  @BeforeEach
  void beforeEach() {
    allocator = new RootAllocator();
  }

  @AfterEach
  void afterEach() {
    allocator.close();
  }

  @Test
  void bindOrder() throws SQLException {
    final Schema schema =
        new Schema(
            Arrays.asList(
                Field.nullable("ints0", new ArrowType.Int(32, true)),
                Field.nullable("ints1", new ArrowType.Int(32, true)),
                Field.nullable("ints2", new ArrowType.Int(32, true))));
    try (final MockPreparedStatement statement = new MockPreparedStatement();
        final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
      final JdbcParameterBinder binder =
          JdbcParameterBinder.builder(statement, root)
              .bind(/*parameterIndex=*/ 1, /*columnIndex=*/ 2)
              .bind(/*parameterIndex=*/ 2, /*columnIndex=*/ 0)
              .build();
      assertThat(binder.next()).isFalse();

      final IntVector ints0 = (IntVector) root.getVector(0);
      final IntVector ints1 = (IntVector) root.getVector(1);
      final IntVector ints2 = (IntVector) root.getVector(2);
      ints0.setSafe(0, 4);
      ints0.setNull(1);
      ints1.setNull(0);
      ints1.setSafe(1, -8);
      ints2.setNull(0);
      ints2.setSafe(1, 12);
      root.setRowCount(2);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isNull();
      assertThat(statement.getParamType(1)).isEqualTo(Types.INTEGER);
      assertThat(statement.getParamValue(2)).isEqualTo(4);
      assertThat(statement.getParam(3)).isNull();
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(12);
      assertThat(statement.getParamValue(2)).isNull();
      assertThat(statement.getParamType(2)).isEqualTo(Types.INTEGER);
      assertThat(statement.getParam(3)).isNull();
      assertThat(binder.next()).isFalse();

      binder.reset();

      ints0.setNull(0);
      ints0.setSafe(1, -2);
      ints2.setNull(0);
      ints2.setSafe(1, 6);
      root.setRowCount(2);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isNull();
      assertThat(statement.getParamType(1)).isEqualTo(Types.INTEGER);
      assertThat(statement.getParamValue(2)).isNull();
      assertThat(statement.getParamType(2)).isEqualTo(Types.INTEGER);
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(6);
      assertThat(statement.getParamValue(2)).isEqualTo(-2);
      assertThat(statement.getParam(3)).isNull();
      assertThat(binder.next()).isFalse();
    }
  }

  @Test
  void customBinder() throws SQLException {
    final Schema schema =
        new Schema(Collections.singletonList(Field.nullable("ints0", new ArrowType.Int(32, true))));

    try (final MockPreparedStatement statement = new MockPreparedStatement();
        final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
      final JdbcParameterBinder binder =
          JdbcParameterBinder.builder(statement, root)
              .bind(
                  /*parameterIndex=*/ 1,
                  new ColumnBinder() {
                    private final IntVector vector = (IntVector) root.getVector(0);

                    @Override
                    public void bind(PreparedStatement statement, int parameterIndex, int rowIndex)
                        throws SQLException {
                      Integer value = vector.getObject(rowIndex);
                      if (value == null) {
                        statement.setString(parameterIndex, "null");
                      } else {
                        statement.setString(parameterIndex, Integer.toString(value));
                      }
                    }

                    @Override
                    public int getJdbcType() {
                      return Types.INTEGER;
                    }

                    @Override
                    public FieldVector getVector() {
                      return vector;
                    }
                  })
              .build();
      assertThat(binder.next()).isFalse();

      final IntVector ints = (IntVector) root.getVector(0);
      ints.setSafe(0, 4);
      ints.setNull(1);

      root.setRowCount(2);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo("4");
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo("null");
      assertThat(binder.next()).isFalse();
    }
  }

  @Test
  void bool() throws SQLException {
    testSimpleType(
        ArrowType.Bool.INSTANCE,
        Types.BOOLEAN,
        (BitVector vector, Integer index, Boolean value) -> vector.setSafe(index, value ? 1 : 0),
        BitVector::setNull,
        Arrays.asList(true, false, true));
  }

  @Test
  void int8() throws SQLException {
    testSimpleType(
        new ArrowType.Int(8, true),
        Types.TINYINT,
        TinyIntVector::setSafe,
        TinyIntVector::setNull,
        Arrays.asList(Byte.MAX_VALUE, Byte.MIN_VALUE, (byte) 42));
  }

  @Test
  void int16() throws SQLException {
    testSimpleType(
        new ArrowType.Int(16, true),
        Types.SMALLINT,
        SmallIntVector::setSafe,
        SmallIntVector::setNull,
        Arrays.asList(Short.MAX_VALUE, Short.MIN_VALUE, (short) 42));
  }

  @Test
  void int32() throws SQLException {
    testSimpleType(
        new ArrowType.Int(32, true),
        Types.INTEGER,
        IntVector::setSafe,
        IntVector::setNull,
        Arrays.asList(Integer.MAX_VALUE, Integer.MIN_VALUE, 42));
  }

  @Test
  void int64() throws SQLException {
    testSimpleType(
        new ArrowType.Int(64, true),
        Types.BIGINT,
        BigIntVector::setSafe,
        BigIntVector::setNull,
        Arrays.asList(Long.MAX_VALUE, Long.MIN_VALUE, 42L));
  }

  @Test
  void float32() throws SQLException {
    testSimpleType(
        new ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE),
        Types.REAL,
        Float4Vector::setSafe,
        Float4Vector::setNull,
        Arrays.asList(Float.MIN_VALUE, Float.MAX_VALUE, Float.POSITIVE_INFINITY));
  }

  @Test
  void float64() throws SQLException {
    testSimpleType(
        new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE),
        Types.DOUBLE,
        Float8Vector::setSafe,
        Float8Vector::setNull,
        Arrays.asList(Double.MIN_VALUE, Double.MAX_VALUE, Double.POSITIVE_INFINITY));
  }

  @Test
  void time32() throws SQLException {
    testSimpleType(
        new ArrowType.Time(TimeUnit.SECOND, 32),
        Types.TIME,
        (valueVectors, index, value) ->
            valueVectors.setSafe(index, (int) (value.getTime() / 1_000)),
        TimeSecVector::setNull,
        Arrays.asList(new Time(-128_000), new Time(104_000), new Time(-42_000)));
    testSimpleType(
        new ArrowType.Time(TimeUnit.MILLISECOND, 32),
        Types.TIME,
        (valueVectors, index, value) -> valueVectors.setSafe(index, (int) value.getTime()),
        TimeMilliVector::setNull,
        Arrays.asList(new Time(-128_000), new Time(104_000), new Time(-42_000)));
  }

  @Test
  void time64() throws SQLException {
    testSimpleType(
        new ArrowType.Time(TimeUnit.MICROSECOND, 64),
        Types.TIME,
        (valueVectors, index, value) -> valueVectors.setSafe(index, (value.getTime() * 1_000)),
        TimeMicroVector::setNull,
        Arrays.asList(new Time(-128_000), new Time(104_000), new Time(-42_000)));
    testSimpleType(
        new ArrowType.Time(TimeUnit.NANOSECOND, 64),
        Types.TIME,
        (valueVectors, index, value) -> valueVectors.setSafe(index, (value.getTime() * 1_000_000)),
        TimeNanoVector::setNull,
        Arrays.asList(new Time(-128), new Time(104), new Time(-42)));
  }

  @Test
  void date32() throws SQLException {
    testSimpleType(
        new ArrowType.Date(DateUnit.DAY),
        Types.DATE,
        (valueVectors, index, value) ->
            valueVectors.setSafe(index, (int) (value.getTime() / MILLIS_PER_DAY)),
        DateDayVector::setNull,
        Arrays.asList(
            new Date(-5 * MILLIS_PER_DAY), new Date(2 * MILLIS_PER_DAY), new Date(MILLIS_PER_DAY)));
  }

  @Test
  void date64() throws SQLException {
    testSimpleType(
        new ArrowType.Date(DateUnit.MILLISECOND),
        Types.DATE,
        (valueVectors, index, value) -> valueVectors.setSafe(index, value.getTime()),
        DateMilliVector::setNull,
        Arrays.asList(
            new Date(-5 * MILLIS_PER_DAY), new Date(2 * MILLIS_PER_DAY), new Date(MILLIS_PER_DAY)));
  }

  @Test
  void timestamp() throws SQLException {
    List<Timestamp> values =
        Arrays.asList(new Timestamp(-128_000), new Timestamp(104_000), new Timestamp(-42_000));
    testSimpleType(
        new ArrowType.Timestamp(TimeUnit.SECOND, null),
        Types.TIMESTAMP,
        (valueVectors, index, value) -> valueVectors.setSafe(index, value.getTime() / 1_000),
        TimeStampSecVector::setNull,
        values);
    testSimpleType(
        new ArrowType.Timestamp(TimeUnit.MILLISECOND, null),
        Types.TIMESTAMP,
        (valueVectors, index, value) -> valueVectors.setSafe(index, value.getTime()),
        TimeStampMilliVector::setNull,
        values);
    testSimpleType(
        new ArrowType.Timestamp(TimeUnit.MICROSECOND, null),
        Types.TIMESTAMP,
        (valueVectors, index, value) -> valueVectors.setSafe(index, value.getTime() * 1_000),
        TimeStampMicroVector::setNull,
        values);
    testSimpleType(
        new ArrowType.Timestamp(TimeUnit.NANOSECOND, null),
        Types.TIMESTAMP,
        (valueVectors, index, value) -> valueVectors.setSafe(index, value.getTime() * 1_000_000),
        TimeStampNanoVector::setNull,
        values);
  }

  @Test
  void timestampTz() throws SQLException {
    List<Timestamp> values =
        Arrays.asList(new Timestamp(-128_000), new Timestamp(104_000), new Timestamp(-42_000));
    testSimpleType(
        new ArrowType.Timestamp(TimeUnit.SECOND, "UTC"),
        Types.TIMESTAMP_WITH_TIMEZONE,
        (valueVectors, index, value) -> valueVectors.setSafe(index, value.getTime() / 1_000),
        TimeStampSecTZVector::setNull,
        values);
    testSimpleType(
        new ArrowType.Timestamp(TimeUnit.MILLISECOND, "UTC"),
        Types.TIMESTAMP_WITH_TIMEZONE,
        (valueVectors, index, value) -> valueVectors.setSafe(index, value.getTime()),
        TimeStampMilliTZVector::setNull,
        values);
    testSimpleType(
        new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC"),
        Types.TIMESTAMP_WITH_TIMEZONE,
        (valueVectors, index, value) -> valueVectors.setSafe(index, value.getTime() * 1_000),
        TimeStampMicroTZVector::setNull,
        values);
    testSimpleType(
        new ArrowType.Timestamp(TimeUnit.NANOSECOND, "UTC"),
        Types.TIMESTAMP_WITH_TIMEZONE,
        (valueVectors, index, value) -> valueVectors.setSafe(index, value.getTime() * 1_000_000),
        TimeStampNanoTZVector::setNull,
        values);
  }

  @Test
  void utf8() throws SQLException {
    testSimpleType(
        ArrowType.Utf8.INSTANCE,
        Types.VARCHAR,
        (VarCharVector vector, Integer index, String value) ->
            vector.setSafe(index, value.getBytes(StandardCharsets.UTF_8)),
        BaseVariableWidthVector::setNull,
        Arrays.asList("", "foobar", "abc"));
  }

  @Test
  void largeUtf8() throws SQLException {
    testSimpleType(
        ArrowType.LargeUtf8.INSTANCE,
        Types.LONGVARCHAR,
        (LargeVarCharVector vector, Integer index, String value) ->
            vector.setSafe(index, value.getBytes(StandardCharsets.UTF_8)),
        BaseLargeVariableWidthVector::setNull,
        Arrays.asList("", "foobar", "abc"));
  }

  @Test
  void binary() throws SQLException {
    testSimpleType(
        ArrowType.Binary.INSTANCE,
        Types.VARBINARY,
        (VarBinaryVector vector, Integer index, byte[] value) -> vector.setSafe(index, value),
        BaseVariableWidthVector::setNull,
        Arrays.asList(new byte[0], new byte[] {2, -4}, new byte[] {0, -1, 127, -128}));
  }

  @Test
  void largeBinary() throws SQLException {
    testSimpleType(
        ArrowType.LargeBinary.INSTANCE,
        Types.LONGVARBINARY,
        (LargeVarBinaryVector vector, Integer index, byte[] value) -> vector.setSafe(index, value),
        BaseLargeVariableWidthVector::setNull,
        Arrays.asList(new byte[0], new byte[] {2, -4}, new byte[] {0, -1, 127, -128}));
  }

  @Test
  void fixedSizeBinary() throws SQLException {
    testSimpleType(
        new ArrowType.FixedSizeBinary(3),
        Types.BINARY,
        FixedSizeBinaryVector::setSafe,
        FixedSizeBinaryVector::setNull,
        Arrays.asList(new byte[3], new byte[] {1, 2, -4}, new byte[] {-1, 127, -128}));
  }

  @Test
  void decimal128() throws SQLException {
    testSimpleType(
        new ArrowType.Decimal(/*precision*/ 12, /*scale*/ 3, 128),
        Types.DECIMAL,
        DecimalVector::setSafe,
        DecimalVector::setNull,
        Arrays.asList(
            new BigDecimal("120.429"), new BigDecimal("-10590.123"), new BigDecimal("0.000")));
  }

  @Test
  void decimal256() throws SQLException {
    testSimpleType(
        new ArrowType.Decimal(/*precision*/ 12, /*scale*/ 3, 256),
        Types.DECIMAL,
        Decimal256Vector::setSafe,
        Decimal256Vector::setNull,
        Arrays.asList(
            new BigDecimal("120.429"), new BigDecimal("-10590.123"), new BigDecimal("0.000")));
  }

  @Test
  void listOfDouble() throws SQLException {
    TriConsumer<ListVector, Integer, Double[]> setValue =
        (listVector, index, values) -> {
          org.apache.arrow.vector.complex.impl.UnionListWriter writer = listVector.getWriter();
          writer.setPosition(index);
          writer.startList();
          Arrays.stream(values).forEach(doubleValue -> writer.float8().writeFloat8(doubleValue));
          writer.endList();
          listVector.setLastSet(index);
        };
    List<Double[]> values =
        Arrays.asList(
            new Double[] {0.0, Math.PI}, new Double[] {1.1, -352346.2, 2355.6},
            new Double[] {-1024.3}, new Double[] {});
    testListType(
        new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE),
        setValue,
        ListVector::setNull,
        values);
  }

  @Test
  void listOfInt64() throws SQLException {
    TriConsumer<ListVector, Integer, Long[]> setValue =
        (listVector, index, values) -> {
          org.apache.arrow.vector.complex.impl.UnionListWriter writer = listVector.getWriter();
          writer.setPosition(index);
          writer.startList();
          Arrays.stream(values).forEach(longValue -> writer.bigInt().writeBigInt(longValue));
          writer.endList();
          listVector.setLastSet(index);
        };
    List<Long[]> values =
        Arrays.asList(
            new Long[] {1L, 2L, 3L},
            new Long[] {4L, 5L},
            new Long[] {512L, 1024L, 2048L, 4096L},
            new Long[] {});
    testListType((ArrowType) new ArrowType.Int(64, true), setValue, ListVector::setNull, values);
  }

  @Test
  void listOfInt32() throws SQLException {
    TriConsumer<ListVector, Integer, Integer[]> setValue =
        (listVector, index, values) -> {
          org.apache.arrow.vector.complex.impl.UnionListWriter writer = listVector.getWriter();
          writer.setPosition(index);
          writer.startList();
          Arrays.stream(values).forEach(integerValue -> writer.integer().writeInt(integerValue));
          writer.endList();
          listVector.setLastSet(index);
        };
    List<Integer[]> values =
        Arrays.asList(
            new Integer[] {1, 2, 3},
            new Integer[] {4, 5},
            new Integer[] {512, 1024, 2048, 4096},
            new Integer[] {});
    testListType((ArrowType) new ArrowType.Int(32, true), setValue, ListVector::setNull, values);
  }

  @Test
  void listOfBoolean() throws SQLException {
    TriConsumer<ListVector, Integer, Boolean[]> setValue =
        (listVector, index, values) -> {
          org.apache.arrow.vector.complex.impl.UnionListWriter writer = listVector.getWriter();
          writer.setPosition(index);
          writer.startList();
          Arrays.stream(values)
              .forEach(booleanValue -> writer.bit().writeBit(booleanValue ? 1 : 0));
          writer.endList();
          listVector.setLastSet(index);
        };
    List<Boolean[]> values =
        Arrays.asList(
            new Boolean[] {true, false},
            new Boolean[] {false, false},
            new Boolean[] {true, true, false, true},
            new Boolean[] {});
    testListType((ArrowType) new ArrowType.Bool(), setValue, ListVector::setNull, values);
  }

  @Test
  void listOfString() throws SQLException {
    TriConsumer<ListVector, Integer, String[]> setValue =
        (listVector, index, values) -> {
          org.apache.arrow.vector.complex.impl.UnionListWriter writer = listVector.getWriter();
          writer.setPosition(index);
          writer.startList();
          Arrays.stream(values)
              .forEach(
                  stringValue -> {
                    if (stringValue != null) {
                      byte[] stringValueBytes = stringValue.getBytes(StandardCharsets.UTF_8);
                      try (ArrowBuf stringBuffer = allocator.buffer(stringValueBytes.length)) {
                        stringBuffer.writeBytes(stringValueBytes);
                        writer.varChar().writeVarChar(0, stringValueBytes.length, stringBuffer);
                      }
                    } else {
                      writer.varChar().writeNull();
                    }
                  });
          writer.endList();
          listVector.setLastSet(index);
        };
    List<String[]> values =
        Arrays.asList(
            new String[] {"aaaa", "b1"},
            new String[] {"c", null, "d"},
            new String[] {"e", "f", "g", "h"},
            new String[] {});
    testListType((ArrowType) new ArrowType.Utf8(), setValue, ListVector::setNull, values);
  }

  @Test
  void mapOfString() throws SQLException {
    TriConsumer<MapVector, Integer, Map<String, String>> setValue =
        (mapVector, index, values) -> {
          org.apache.arrow.vector.complex.impl.UnionMapWriter mapWriter = mapVector.getWriter();
          mapWriter.setPosition(index);
          mapWriter.startMap();
          values
              .entrySet()
              .forEach(
                  mapValue -> {
                    if (mapValue != null) {
                      byte[] keyBytes = mapValue.getKey().getBytes(StandardCharsets.UTF_8);
                      byte[] valueBytes = mapValue.getValue().getBytes(StandardCharsets.UTF_8);
                      try (ArrowBuf keyBuf = allocator.buffer(keyBytes.length);
                          ArrowBuf valueBuf = allocator.buffer(valueBytes.length); ) {
                        mapWriter.startEntry();
                        keyBuf.writeBytes(keyBytes);
                        valueBuf.writeBytes(valueBytes);
                        mapWriter.key().varChar().writeVarChar(0, keyBytes.length, keyBuf);
                        mapWriter.value().varChar().writeVarChar(0, valueBytes.length, valueBuf);
                        mapWriter.endEntry();
                      }
                    } else {
                      mapWriter.writeNull();
                    }
                  });
          mapWriter.endMap();
        };

    JsonStringHashMap<String, String> value1 = new JsonStringHashMap<String, String>();
    value1.put("a", "b");
    value1.put("c", "d");
    JsonStringHashMap<String, String> value2 = new JsonStringHashMap<String, String>();
    value2.put("d", "e");
    value2.put("f", "g");
    value2.put("k", "l");
    JsonStringHashMap<String, String> value3 = new JsonStringHashMap<String, String>();
    value3.put("y", "z");
    value3.put("arrow", "cool");
    List<Map<String, String>> values =
        Arrays.asList(value1, value2, value3, Collections.emptyMap());
    testMapType(
        new ArrowType.Map(true), setValue, MapVector::setNull, values, new ArrowType.Utf8());
  }

  @Test
  void mapOfInteger() throws SQLException {
    TriConsumer<MapVector, Integer, Map<Integer, Integer>> setValue =
        (mapVector, index, values) -> {
          org.apache.arrow.vector.complex.impl.UnionMapWriter mapWriter = mapVector.getWriter();
          mapWriter.setPosition(index);
          mapWriter.startMap();
          values
              .entrySet()
              .forEach(
                  mapValue -> {
                    if (mapValue != null) {
                      mapWriter.startEntry();
                      mapWriter.key().integer().writeInt(mapValue.getKey());
                      mapWriter.value().integer().writeInt(mapValue.getValue());
                      mapWriter.endEntry();
                    } else {
                      mapWriter.writeNull();
                    }
                  });
          mapWriter.endMap();
        };

    JsonStringHashMap<Integer, Integer> value1 = new JsonStringHashMap<Integer, Integer>();
    value1.put(1, 2);
    value1.put(3, 4);
    JsonStringHashMap<Integer, Integer> value2 = new JsonStringHashMap<Integer, Integer>();
    value2.put(5, 6);
    value2.put(7, 8);
    value2.put(9, 1024);
    JsonStringHashMap<Integer, Integer> value3 = new JsonStringHashMap<Integer, Integer>();
    value3.put(Integer.MIN_VALUE, Integer.MAX_VALUE);
    value3.put(0, 4096);
    List<Map<Integer, Integer>> values =
        Arrays.asList(value1, value2, value3, Collections.emptyMap());
    testMapType(
        new ArrowType.Map(true), setValue, MapVector::setNull, values, new ArrowType.Int(32, true));
  }

  @FunctionalInterface
  interface TriConsumer<T, U, V> {
    void accept(T value1, U value2, V value3);
  }

  <T, V extends FieldVector> void testSimpleType(
      ArrowType arrowType,
      int jdbcType,
      TriConsumer<V, Integer, T> setValue,
      BiConsumer<V, Integer> setNull,
      List<T> values)
      throws SQLException {
    Schema schema = new Schema(Collections.singletonList(Field.nullable("field", arrowType)));
    try (final MockPreparedStatement statement = new MockPreparedStatement();
        final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
      final JdbcParameterBinder binder =
          JdbcParameterBinder.builder(statement, root).bindAll().build();
      assertThat(binder.next()).isFalse();

      @SuppressWarnings("unchecked")
      final V vector = (V) root.getVector(0);
      final ColumnBinder columnBinder = ColumnBinder.forVector(vector);
      assertThat(columnBinder.getJdbcType()).isEqualTo(jdbcType);

      setValue.accept(vector, 0, values.get(0));
      setValue.accept(vector, 1, values.get(1));
      setNull.accept(vector, 2);
      root.setRowCount(3);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isNull();
      assertThat(statement.getParamType(1)).isEqualTo(jdbcType);
      assertThat(binder.next()).isFalse();

      binder.reset();

      setNull.accept(vector, 0);
      setValue.accept(vector, 1, values.get(2));
      setValue.accept(vector, 2, values.get(0));
      setValue.accept(vector, 3, values.get(2));
      setValue.accept(vector, 4, values.get(1));
      root.setRowCount(5);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isNull();
      assertThat(statement.getParamType(1)).isEqualTo(jdbcType);
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(2));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(2));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1));
      assertThat(binder.next()).isFalse();
    }

    // Non-nullable (since some types have a specialized binder)
    schema = new Schema(Collections.singletonList(Field.notNullable("field", arrowType)));
    try (final MockPreparedStatement statement = new MockPreparedStatement();
        final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
      final JdbcParameterBinder binder =
          JdbcParameterBinder.builder(statement, root).bindAll().build();
      assertThat(binder.next()).isFalse();

      @SuppressWarnings("unchecked")
      final V vector = (V) root.getVector(0);
      setValue.accept(vector, 0, values.get(0));
      setValue.accept(vector, 1, values.get(1));
      root.setRowCount(2);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1));
      assertThat(binder.next()).isFalse();

      binder.reset();

      setValue.accept(vector, 0, values.get(0));
      setValue.accept(vector, 1, values.get(2));
      setValue.accept(vector, 2, values.get(0));
      setValue.accept(vector, 3, values.get(2));
      setValue.accept(vector, 4, values.get(1));
      root.setRowCount(5);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(2));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(2));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1));
      assertThat(binder.next()).isFalse();
    }
  }

  <T, V extends FieldVector> void testListType(
      ArrowType arrowType,
      TriConsumer<V, Integer, T> setValue,
      BiConsumer<V, Integer> setNull,
      List<T> values)
      throws SQLException {
    int jdbcType = Types.ARRAY;
    Schema schema =
        new Schema(
            Collections.singletonList(
                new Field(
                    "field",
                    FieldType.nullable(new ArrowType.List()),
                    Collections.singletonList(
                        new Field("element", FieldType.notNullable(arrowType), null)))));
    try (final MockPreparedStatement statement = new MockPreparedStatement();
        final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
      final JdbcParameterBinder binder =
          JdbcParameterBinder.builder(statement, root).bindAll().build();
      assertThat(binder.next()).isFalse();

      @SuppressWarnings("unchecked")
      final V vector = (V) root.getVector(0);
      final ColumnBinder columnBinder = ColumnBinder.forVector(vector);
      assertThat(columnBinder.getJdbcType()).isEqualTo(jdbcType);

      setValue.accept(vector, 0, values.get(0));
      setValue.accept(vector, 1, values.get(1));
      setNull.accept(vector, 2);
      root.setRowCount(3);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isNull();
      assertThat(statement.getParamType(1)).isEqualTo(jdbcType);
      assertThat(binder.next()).isFalse();

      binder.reset();

      setNull.accept(vector, 0);
      setValue.accept(vector, 1, values.get(3));
      setValue.accept(vector, 2, values.get(0));
      setValue.accept(vector, 3, values.get(2));
      setValue.accept(vector, 4, values.get(1));
      root.setRowCount(5);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isNull();
      assertThat(statement.getParamType(1)).isEqualTo(jdbcType);
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(3));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(2));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1));
      assertThat(binder.next()).isFalse();
    }

    // Non-nullable (since some types have a specialized binder)
    schema =
        new Schema(
            Collections.singletonList(
                new Field(
                    "field",
                    FieldType.notNullable(new ArrowType.List()),
                    Collections.singletonList(
                        new Field("element", FieldType.notNullable(arrowType), null)))));
    try (final MockPreparedStatement statement = new MockPreparedStatement();
        final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
      final JdbcParameterBinder binder =
          JdbcParameterBinder.builder(statement, root).bindAll().build();
      assertThat(binder.next()).isFalse();

      @SuppressWarnings("unchecked")
      final V vector = (V) root.getVector(0);
      setValue.accept(vector, 0, values.get(0));
      setValue.accept(vector, 1, values.get(1));
      root.setRowCount(2);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1));
      assertThat(binder.next()).isFalse();

      binder.reset();

      setValue.accept(vector, 0, values.get(0));
      setValue.accept(vector, 1, values.get(2));
      setValue.accept(vector, 2, values.get(0));
      setValue.accept(vector, 3, values.get(2));
      setValue.accept(vector, 4, values.get(1));
      root.setRowCount(5);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(2));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(2));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1));
      assertThat(binder.next()).isFalse();
    }
  }

  <T, V extends FieldVector> void testMapType(
      ArrowType arrowType,
      TriConsumer<V, Integer, T> setValue,
      BiConsumer<V, Integer> setNull,
      List<T> values,
      ArrowType elementType)
      throws SQLException {
    int jdbcType = Types.VARCHAR;
    FieldType keyType = new FieldType(false, elementType, null, null);
    FieldType mapType = new FieldType(false, ArrowType.Struct.INSTANCE, null, null);
    Schema schema =
        new Schema(
            Collections.singletonList(
                new Field(
                    "field",
                    FieldType.nullable(arrowType),
                    Collections.singletonList(
                        new Field(
                            MapVector.KEY_NAME,
                            mapType,
                            Arrays.asList(
                                new Field(MapVector.KEY_NAME, keyType, null),
                                new Field(MapVector.VALUE_NAME, keyType, null)))))));
    try (final MockPreparedStatement statement = new MockPreparedStatement();
        final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
      final JdbcParameterBinder binder =
          JdbcParameterBinder.builder(statement, root).bindAll().build();
      assertThat(binder.next()).isFalse();

      @SuppressWarnings("unchecked")
      final V vector = (V) root.getVector(0);
      final ColumnBinder columnBinder = ColumnBinder.forVector(vector);
      assertThat(columnBinder.getJdbcType()).isEqualTo(jdbcType);

      setValue.accept(vector, 0, values.get(0));
      setValue.accept(vector, 1, values.get(1));
      setNull.accept(vector, 2);
      root.setRowCount(3);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0).toString());
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1).toString());
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isNull();
      assertThat(statement.getParamType(1)).isEqualTo(jdbcType);
      assertThat(binder.next()).isFalse();

      binder.reset();

      setNull.accept(vector, 0);
      setValue.accept(vector, 1, values.get(3));
      setValue.accept(vector, 2, values.get(0));
      setValue.accept(vector, 3, values.get(2));
      setValue.accept(vector, 4, values.get(1));
      root.setRowCount(5);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isNull();
      assertThat(statement.getParamType(1)).isEqualTo(jdbcType);
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(3).toString());
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0).toString());
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(2).toString());
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1).toString());
      assertThat(binder.next()).isFalse();
    }

    // Non-nullable (since some types have a specialized binder)
    schema =
        new Schema(
            Collections.singletonList(
                new Field(
                    "field",
                    FieldType.notNullable(arrowType),
                    Collections.singletonList(
                        new Field(
                            MapVector.KEY_NAME,
                            mapType,
                            Arrays.asList(
                                new Field(MapVector.KEY_NAME, keyType, null),
                                new Field(MapVector.VALUE_NAME, keyType, null)))))));
    try (final MockPreparedStatement statement = new MockPreparedStatement();
        final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
      @SuppressWarnings("unchecked")
      final V vector = (V) root.getVector(0);

      final JdbcParameterBinder binder =
          JdbcParameterBinder.builder(statement, root)
              .bind(
                  1,
                  new org.apache.arrow.adapter.jdbc.binder.MapBinder(
                      (MapVector) vector, Types.OTHER))
              .build();
      assertThat(binder.next()).isFalse();

      setValue.accept(vector, 0, values.get(0));
      setValue.accept(vector, 1, values.get(1));
      root.setRowCount(2);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1));
      assertThat(binder.next()).isFalse();

      binder.reset();

      setValue.accept(vector, 0, values.get(0));
      setValue.accept(vector, 1, values.get(2));
      setValue.accept(vector, 2, values.get(0));
      setValue.accept(vector, 3, values.get(2));
      setValue.accept(vector, 4, values.get(1));
      root.setRowCount(5);

      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(2));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(0));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(2));
      assertThat(binder.next()).isTrue();
      assertThat(statement.getParamValue(1)).isEqualTo(values.get(1));
      assertThat(binder.next()).isFalse();
    }
  }
}
