public static <T> Write<T> write(String jdbcUrl, String table) {
    return new AutoValue_ClickHouseIO_Write.Builder<T>()
        .jdbcUrl(jdbcUrl)
        .table(table)
        .properties(new Properties())
        .maxInsertBlockSize(DEFAULT_MAX_INSERT_BLOCK_SIZE)
        .initialBackoff(DEFAULT_INITIAL_BACKOFF)
        .maxRetries(DEFAULT_MAX_RETRIES)
        .maxCumulativeBackoff(DEFAULT_MAX_CUMULATIVE_BACKOFF)
        .build()
        .withInsertDeduplicate(true)
        .withInsertDistributedSync(true);
  }
@Test
  public void testArrayOfPrimitiveTypes() throws Exception {
    Schema schema =
        Schema.of(
            Schema.Field.of("f0", FieldType.array(FieldType.DATETIME)),
            Schema.Field.of("f1", FieldType.array(FieldType.DATETIME)),
            Schema.Field.of("f2", FieldType.array(FieldType.FLOAT)),
            Schema.Field.of("f3", FieldType.array(FieldType.DOUBLE)),
            Schema.Field.of("f4", FieldType.array(FieldType.BYTE)),
            Schema.Field.of("f5", FieldType.array(FieldType.INT16)),
            Schema.Field.of("f6", FieldType.array(FieldType.INT32)),
            Schema.Field.of("f7", FieldType.array(FieldType.INT64)),
            Schema.Field.of("f8", FieldType.array(FieldType.STRING)),
            Schema.Field.of("f9", FieldType.array(FieldType.INT16)),
            Schema.Field.of("f10", FieldType.array(FieldType.INT32)),
            Schema.Field.of("f11", FieldType.array(FieldType.INT64)),
            Schema.Field.of("f12", FieldType.array(FieldType.INT64)),
            Schema.Field.of("f13", FieldType.array(FieldType.STRING)),
            Schema.Field.of("f14", FieldType.array(FieldType.STRING)),
            Schema.Field.of("f15", FieldType.array(FieldType.BOOLEAN)));
    Row row1 =
        Row.withSchema(schema)
            .addArray(
                new DateTime(2030, 10, 1, 0, 0, 0, DateTimeZone.UTC),
                new DateTime(2031, 10, 1, 0, 0, 0, DateTimeZone.UTC))
            .addArray(
                new DateTime(2030, 10, 9, 8, 7, 6, DateTimeZone.UTC),
                new DateTime(2031, 10, 9, 8, 7, 6, DateTimeZone.UTC))
            .addArray(2.2f, 3.3f)
            .addArray(3.3, 4.4)
            .addArray((byte) 4, (byte) 5)
            .addArray((short) 5, (short) 6)
            .addArray(6, 7)
            .addArray(7L, 8L)
            .addArray("eight", "nine")
            .addArray((short) 9, (short) 10)
            .addArray(10, 11)
            .addArray(11L, 12L)
            .addArray(12L, 13L)
            .addArray("abc", "cde")
            .addArray("cde", "abc")
            .addArray(true, false)
            .build();

    executeSql(
        "CREATE TABLE test_array_of_primitive_types ("
            + "f0  Array(Date),"
            + "f1  Array(DateTime),"
            + "f2  Array(Float32),"
            + "f3  Array(Float64),"
            + "f4  Array(Int8),"
            + "f5  Array(Int16),"
            + "f6  Array(Int32),"
            + "f7  Array(Int64),"
            + "f8  Array(String),"
            + "f9  Array(UInt8),"
            + "f10 Array(UInt16),"
            + "f11 Array(UInt32),"
            + "f12 Array(UInt64),"
            + "f13 Array(Enum8('abc' = 1, 'cde' = 2)),"
            + "f14 Array(Enum16('abc' = -1, 'cde' = -2)),"
            + "f15 Array(Bool)"
            + ") ENGINE=Log");

    pipeline
        .apply(Create.of(row1).withRowSchema(schema))
        .apply(write("test_array_of_primitive_types"));

    pipeline.run().waitUntilFinish();

    try (ResultSet rs = executeQuery("SELECT * FROM test_array_of_primitive_types")) {
      rs.next();

      assertEquals("[2030-10-01, 2031-10-01]", rs.getString("f0"));
      assertEquals("[2030-10-09T08:07:06, 2031-10-09T08:07:06]", rs.getString("f1"));
      // Since comparing float/double values is not precise, we compare the string representation
      assertEquals("[2.2,3.3]", rs.getString("f2"));
      assertEquals("[3.3,4.4]", rs.getString("f3"));
      assertArrayEquals(new byte[] {4, 5}, (byte[]) rs.getArray("f4").getArray());
      assertArrayEquals(new short[] {5, 6}, (short[]) rs.getArray("f5").getArray());
      assertArrayEquals(new int[] {6, 7}, (int[]) rs.getArray("f6").getArray());
      assertArrayEquals(new long[] {7L, 8L}, (long[]) rs.getArray("f7").getArray());
      assertArrayEquals(new String[] {"eight", "nine"}, (String[]) rs.getArray("f8").getArray());
      assertArrayEquals(new byte[] {9, 10}, (byte[]) rs.getArray("f9").getArray());
      assertArrayEquals(new short[] {10, 11}, (short[]) rs.getArray("f10").getArray());
      assertArrayEquals(new int[] {11, 12}, (int[]) rs.getArray("f11").getArray());
      assertArrayEquals(new long[] {12L, 13L}, (long[]) rs.getArray("f12").getArray());
      assertArrayEquals(new String[] {"abc", "cde"}, (String[]) rs.getArray("f13").getArray());
      assertArrayEquals(new String[] {"cde", "abc"}, (String[]) rs.getArray("f14").getArray());
      assertArrayEquals(new boolean[] {true, false}, (boolean[]) rs.getArray("f15").getArray());
    }
  }