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 testPrimitiveTypes() throws Exception {
    Schema schema =
        Schema.of(
            Schema.Field.of("f0", FieldType.DATETIME),
            Schema.Field.of("f1", FieldType.DATETIME),
            Schema.Field.of("f2", FieldType.FLOAT),
            Schema.Field.of("f3", FieldType.DOUBLE),
            Schema.Field.of("f4", FieldType.BYTE),
            Schema.Field.of("f5", FieldType.INT16),
            Schema.Field.of("f6", FieldType.INT32),
            Schema.Field.of("f7", FieldType.INT64),
            Schema.Field.of("f8", FieldType.STRING),
            Schema.Field.of("f9", FieldType.INT16),
            Schema.Field.of("f10", FieldType.INT32),
            Schema.Field.of("f11", FieldType.INT64),
            Schema.Field.of("f12", FieldType.INT64),
            Schema.Field.of("f13", FieldType.STRING),
            Schema.Field.of("f14", FieldType.STRING),
            Schema.Field.of("f15", FieldType.STRING),
            Schema.Field.of("f16", FieldType.BYTES),
            Schema.Field.of("f17", FieldType.logicalType(FixedBytes.of(3))),
            Schema.Field.of("f18", FieldType.BOOLEAN),
            Schema.Field.of("f19", FieldType.STRING));
    Row row1 =
        Row.withSchema(schema)
            .addValue(new DateTime(2030, 10, 1, 0, 0, 0, DateTimeZone.UTC))
            .addValue(new DateTime(2030, 10, 9, 8, 7, 6, DateTimeZone.UTC))
            .addValue(2.2f)
            .addValue(3.3)
            .addValue((byte) 4)
            .addValue((short) 5)
            .addValue(6)
            .addValue(7L)
            .addValue("eight")
            .addValue((short) 9)
            .addValue(10)
            .addValue(11L)
            .addValue(12L)
            .addValue("abc")
            .addValue("cde")
            .addValue("qwe")
            .addValue(new byte[] {'a', 's', 'd'})
            .addValue(new byte[] {'z', 'x', 'c'})
            .addValue(true)
            .addValue("lowcardenality")
            .build();

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

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

    pipeline.run().waitUntilFinish();

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

      assertEquals("2030-10-01", rs.getString("f0"));
      assertEquals("2030-10-09 08:07:06", rs.getString("f1"));
      assertEquals("2.2", rs.getString("f2"));
      assertEquals("3.3", rs.getString("f3"));
      assertEquals("4", rs.getString("f4"));
      assertEquals("5", rs.getString("f5"));
      assertEquals("6", rs.getString("f6"));
      assertEquals("7", rs.getString("f7"));
      assertEquals("eight", rs.getString("f8"));
      assertEquals("9", rs.getString("f9"));
      assertEquals("10", rs.getString("f10"));
      assertEquals("11", rs.getString("f11"));
      assertEquals("12", rs.getString("f12"));
      assertEquals("abc", rs.getString("f13"));
      assertEquals("cde", rs.getString("f14"));
      assertArrayEquals(new byte[] {'q', 'w', 'e'}, rs.getBytes("f15"));
      assertArrayEquals(new byte[] {'a', 's', 'd'}, rs.getBytes("f16"));
      assertArrayEquals(new byte[] {'z', 'x', 'c'}, rs.getBytes("f17"));
      assertEquals("true", rs.getString("f18"));
      assertEquals("lowcardenality", rs.getString("f19"));
    }
  }