/*
 * 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 com.datastax.driver.core;

import static com.datastax.driver.core.Assertions.assertThat;
import static com.datastax.driver.core.ConditionChecker.check;
import static com.datastax.driver.core.Metadata.quote;
import static java.util.concurrent.TimeUnit.MINUTES;

import com.datastax.driver.core.utils.CassandraVersion;
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Callable;
import org.testng.annotations.Test;

@CassandraVersion("2.1.0")
public class UserTypesTest extends CCMTestsSupport {

  private List<DataType> DATA_TYPE_PRIMITIVES;
  private Map<DataType, Object> samples;

  private static final List<DataType.Name> DATA_TYPE_NON_PRIMITIVE_NAMES =
      new ArrayList<DataType.Name>(
          EnumSet.of(
              DataType.Name.LIST, DataType.Name.SET, DataType.Name.MAP, DataType.Name.TUPLE));

  private final Callable<Boolean> userTableExists =
      new Callable<Boolean>() {
        @Override
        public Boolean call() throws Exception {
          return cluster().getMetadata().getKeyspace(keyspace).getTable("user") != null;
        }
      };

  @Override
  public void onTestContextInitialized() {
    ProtocolVersion protocolVersion = ccm().getProtocolVersion();
    DATA_TYPE_PRIMITIVES = new ArrayList<DataType>(TestUtils.allPrimitiveTypes(protocolVersion));
    DATA_TYPE_PRIMITIVES.remove(DataType.counter());
    samples = PrimitiveTypeSamples.samples(protocolVersion);
    String type1 = "CREATE TYPE phone (alias text, number text)";
    String type2 =
        "CREATE TYPE \"\"\"User Address\"\"\" (street text, \"ZIP\"\"\" int, phones set<frozen<phone>>)";
    String type3 = "CREATE TYPE type_for_frozen_test(i int)";
    String table = "CREATE TABLE user (id int PRIMARY KEY, addr frozen<\"\"\"User Address\"\"\">)";
    execute(type1, type2, type3, table);
    // Ci tests fail with "unconfigured columnfamily user"
    check().that(userTableExists).before(5, MINUTES).becomesTrue();
  }

  @Test(groups = "short")
  public void should_store_and_retrieve_with_prepared_statements() throws Exception {
    int userId = 0;
    PreparedStatement ins = session().prepare("INSERT INTO user(id, addr) VALUES (?, ?)");
    PreparedStatement sel = session().prepare("SELECT * FROM user WHERE id=?");

    UserType addrDef =
        cluster().getMetadata().getKeyspace(keyspace).getUserType(quote("\"User Address\""));
    UserType phoneDef = cluster().getMetadata().getKeyspace(keyspace).getUserType("phone");

    UDTValue phone1 =
        phoneDef.newValue().setString("alias", "home").setString("number", "0123548790");
    UDTValue phone2 =
        phoneDef.newValue().setString("alias", "work").setString("number", "0698265251");

    UDTValue addr =
        addrDef
            .newValue()
            .setString("street", "1600 Pennsylvania Ave NW")
            .setInt(quote("ZIP\""), 20500)
            .setSet("phones", ImmutableSet.of(phone1, phone2));

    session().execute(ins.bind(userId, addr));

    Row r = session().execute(sel.bind(userId)).one();

    assertThat(r.getInt("id")).isEqualTo(0);
  }

  @Test(groups = "short")
  public void should_store_and_retrieve_with_simple_statements() throws Exception {
    int userId = 1;
    UserType addrDef =
        cluster().getMetadata().getKeyspace(keyspace).getUserType(quote("\"User Address\""));
    UserType phoneDef = cluster().getMetadata().getKeyspace(keyspace).getUserType("phone");

    UDTValue phone1 =
        phoneDef.newValue().setString("alias", "home").setString("number", "0123548790");
    UDTValue phone2 =
        phoneDef.newValue().setString("alias", "work").setString("number", "0698265251");

    UDTValue addr =
        addrDef
            .newValue()
            .setString("street", "1600 Pennsylvania Ave NW")
            .setInt(quote("ZIP\""), 20500)
            .setSet("phones", ImmutableSet.of(phone1, phone2));

    session().execute("INSERT INTO user(id, addr) VALUES (?, ?)", userId, addr);

    Row r = session().execute("SELECT * FROM user WHERE id=?", userId).one();

    assertThat(r.getInt("id")).isEqualTo(userId);
    assertThat(r.getUDTValue("addr")).isEqualTo(addr);
  }

  @Test(groups = "short")
  public void should_store_type_definitions_in_their_keyspace() throws Exception {
    KeyspaceMetadata thisKeyspace = cluster().getMetadata().getKeyspace(this.keyspace);

    // Types that don't exist don't have definitions
    assertThat(thisKeyspace.getUserType("address1")).isNull();
    assertThat(thisKeyspace.getUserType("phone1")).isNull();

    // Types created by this test have definitions
    assertThat(thisKeyspace.getUserType(quote("\"User Address\""))).isNotNull();
    assertThat(thisKeyspace.getUserType("phone")).isNotNull();

    // If we create another keyspace, it doesn't have the definitions of this keyspace
    String otherKeyspaceName = this.keyspace + "_nonEx";
    session()
        .execute(
            "CREATE KEYSPACE "
                + otherKeyspaceName
                + " "
                + "WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor': '1'}");

    KeyspaceMetadata otherKeyspace = cluster().getMetadata().getKeyspace(otherKeyspaceName);
    assertThat(otherKeyspace.getUserType(quote("\"User Address\""))).isNull();
    assertThat(otherKeyspace.getUserType("phone")).isNull();
  }

  @Test(groups = "short")
  public void should_handle_UDT_with_many_fields() throws Exception {
    int MAX_TEST_LENGTH = 1024;

    // create the seed udt
    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < MAX_TEST_LENGTH; ++i) {
      sb.append(String.format("v_%s int", i));

      if (i + 1 < MAX_TEST_LENGTH) sb.append(",");
    }
    session().execute(String.format("CREATE TYPE lengthy_udt (%s)", sb.toString()));

    // create a table with multiple sizes of udts
    session().execute("CREATE TABLE lengthy_udt_table (k int PRIMARY KEY, v frozen<lengthy_udt>)");

    // hold onto the UserType for future use
    UserType udtDef = cluster().getMetadata().getKeyspace(keyspace).getUserType("lengthy_udt");

    // verify inserts and reads
    for (int i : Arrays.asList(0, 1, 2, 3, MAX_TEST_LENGTH)) {
      // create udt
      UDTValue createdUDT = udtDef.newValue();
      for (int j = 0; j < i; ++j) {
        createdUDT.setInt(j, j);
      }

      // write udt
      session().execute("INSERT INTO lengthy_udt_table (k, v) VALUES (0, ?)", createdUDT);

      // verify udt was written and read correctly
      UDTValue r =
          session().execute("SELECT v FROM lengthy_udt_table WHERE k=0").one().getUDTValue("v");
      assertThat(r.toString()).isEqualTo(createdUDT.toString());
    }
  }

  @Test(groups = "short")
  public void should_store_and_retrieve_UDT_containing_any_primitive_type() throws Exception {
    // create UDT
    List<String> alpha_type_list = new ArrayList<String>();
    int startIndex = (int) 'a';
    for (int i = 0; i < DATA_TYPE_PRIMITIVES.size(); i++) {
      alpha_type_list.add(
          String.format(
              "%s %s",
              Character.toString((char) (startIndex + i)), DATA_TYPE_PRIMITIVES.get(i).getName()));
    }

    session()
        .execute(
            String.format("CREATE TYPE alldatatypes (%s)", Joiner.on(',').join(alpha_type_list)));
    session()
        .execute("CREATE TABLE alldatatypes_table (a int PRIMARY KEY, b frozen<alldatatypes>)");

    // insert UDT data
    UserType alldatatypesDef =
        cluster().getMetadata().getKeyspace(keyspace).getUserType("alldatatypes");
    UDTValue alldatatypes = alldatatypesDef.newValue();

    for (int i = 0; i < DATA_TYPE_PRIMITIVES.size(); i++) {
      DataType dataType = DATA_TYPE_PRIMITIVES.get(i);
      String index = Character.toString((char) (startIndex + i));
      Object sampleData = samples.get(dataType);

      switch (dataType.getName()) {
        case ASCII:
          alldatatypes.setString(index, (String) sampleData);
          break;
        case BIGINT:
          alldatatypes.setLong(index, (Long) sampleData);
          break;
        case BLOB:
          alldatatypes.setBytes(index, (ByteBuffer) sampleData);
          break;
        case BOOLEAN:
          alldatatypes.setBool(index, (Boolean) sampleData);
          break;
        case DECIMAL:
          alldatatypes.setDecimal(index, (BigDecimal) sampleData);
          break;
        case DOUBLE:
          alldatatypes.setDouble(index, (Double) sampleData);
          break;
        case DURATION:
          alldatatypes.set(index, Duration.from(sampleData.toString()), Duration.class);
          break;
        case FLOAT:
          alldatatypes.setFloat(index, (Float) sampleData);
          break;
        case INET:
          alldatatypes.setInet(index, (InetAddress) sampleData);
          break;
        case TINYINT:
          alldatatypes.setByte(index, (Byte) sampleData);
          break;
        case SMALLINT:
          alldatatypes.setShort(index, (Short) sampleData);
          break;
        case INT:
          alldatatypes.setInt(index, (Integer) sampleData);
          break;
        case TEXT:
          alldatatypes.setString(index, (String) sampleData);
          break;
        case TIMESTAMP:
          alldatatypes.setTimestamp(index, ((Date) sampleData));
          break;
        case DATE:
          alldatatypes.setDate(index, ((LocalDate) sampleData));
          break;
        case TIME:
          alldatatypes.setTime(index, ((Long) sampleData));
          break;
        case TIMEUUID:
          alldatatypes.setUUID(index, (UUID) sampleData);
          break;
        case UUID:
          alldatatypes.setUUID(index, (UUID) sampleData);
          break;
        case VARCHAR:
          alldatatypes.setString(index, (String) sampleData);
          break;
        case VARINT:
          alldatatypes.setVarint(index, (BigInteger) sampleData);
          break;
      }
    }

    PreparedStatement ins =
        session().prepare("INSERT INTO alldatatypes_table (a, b) VALUES (?, ?)");
    session().execute(ins.bind(0, alldatatypes));

    // retrieve and verify data
    ResultSet rs = session().execute("SELECT * FROM alldatatypes_table");
    List<Row> rows = rs.all();
    assertThat(rows.size()).isEqualTo(1);

    Row row = rows.get(0);

    assertThat(row.getInt("a")).isEqualTo(0);
    assertThat(row.getUDTValue("b")).isEqualTo(alldatatypes);
  }

  @Test(groups = "short")
  public void should_store_and_retrieve_UDT_containing_collections_and_tuples() throws Exception {
    // counters and durations are not allowed inside collections
    DATA_TYPE_PRIMITIVES.remove(DataType.counter());
    DATA_TYPE_PRIMITIVES.remove(DataType.duration());

    // create UDT
    List<String> alpha_type_list = new ArrayList<String>();
    int startIndex = (int) 'a';
    for (int i = 0; i < DATA_TYPE_NON_PRIMITIVE_NAMES.size(); i++)
      for (int j = 0; j < DATA_TYPE_PRIMITIVES.size(); j++) {
        String typeString;
        if (DATA_TYPE_NON_PRIMITIVE_NAMES.get(i) == DataType.Name.MAP) {
          typeString =
              (String.format(
                  "%s_%s %s<%s, %s>",
                  Character.toString((char) (startIndex + i)),
                  Character.toString((char) (startIndex + j)),
                  DATA_TYPE_NON_PRIMITIVE_NAMES.get(i),
                  DATA_TYPE_PRIMITIVES.get(j).getName(),
                  DATA_TYPE_PRIMITIVES.get(j).getName()));
        } else if (DATA_TYPE_NON_PRIMITIVE_NAMES.get(i) == DataType.Name.TUPLE) {
          typeString =
              (String.format(
                  "%s_%s frozen<%s<%s>>",
                  Character.toString((char) (startIndex + i)),
                  Character.toString((char) (startIndex + j)),
                  DATA_TYPE_NON_PRIMITIVE_NAMES.get(i),
                  DATA_TYPE_PRIMITIVES.get(j).getName()));
        } else {
          typeString =
              (String.format(
                  "%s_%s %s<%s>",
                  Character.toString((char) (startIndex + i)),
                  Character.toString((char) (startIndex + j)),
                  DATA_TYPE_NON_PRIMITIVE_NAMES.get(i),
                  DATA_TYPE_PRIMITIVES.get(j).getName()));
        }
        alpha_type_list.add(typeString);
      }

    session()
        .execute(
            String.format(
                "CREATE TYPE allcollectiontypes (%s)", Joiner.on(',').join(alpha_type_list)));
    session()
        .execute(
            "CREATE TABLE allcollectiontypes_table (a int PRIMARY KEY, b frozen<allcollectiontypes>)");

    // insert UDT data
    UserType allcollectiontypesDef =
        cluster().getMetadata().getKeyspace(keyspace).getUserType("allcollectiontypes");
    UDTValue allcollectiontypes = allcollectiontypesDef.newValue();

    for (int i = 0; i < DATA_TYPE_NON_PRIMITIVE_NAMES.size(); i++)
      for (int j = 0; j < DATA_TYPE_PRIMITIVES.size(); j++) {
        DataType.Name name = DATA_TYPE_NON_PRIMITIVE_NAMES.get(i);
        DataType dataType = DATA_TYPE_PRIMITIVES.get(j);

        String index =
            Character.toString((char) (startIndex + i))
                + "_"
                + Character.toString((char) (startIndex + j));
        Object sampleElement = samples.get(dataType);
        switch (name) {
          case LIST:
            allcollectiontypes.setList(index, Lists.newArrayList(sampleElement));
            break;
          case SET:
            allcollectiontypes.setSet(index, Sets.newHashSet(sampleElement));
            break;
          case MAP:
            allcollectiontypes.setMap(index, ImmutableMap.of(sampleElement, sampleElement));
            break;
          case TUPLE:
            allcollectiontypes.setTupleValue(
                index, cluster().getMetadata().newTupleType(dataType).newValue(sampleElement));
        }
      }

    PreparedStatement ins =
        session().prepare("INSERT INTO allcollectiontypes_table (a, b) VALUES (?, ?)");
    session().execute(ins.bind(0, allcollectiontypes));

    // retrieve and verify data
    ResultSet rs = session().execute("SELECT * FROM allcollectiontypes_table");
    List<Row> rows = rs.all();
    assertThat(rows.size()).isEqualTo(1);

    Row row = rows.get(0);

    assertThat(row.getInt("a")).isEqualTo(0);
    assertThat(row.getUDTValue("b")).isEqualTo(allcollectiontypes);
  }

  @Test(groups = "short")
  public void should_save_and_retrieve_nested_UDTs() throws Exception {
    final int MAX_NESTING_DEPTH = 4;

    // create UDT
    session().execute("CREATE TYPE depth_0 (age int, name text)");

    for (int i = 1; i <= MAX_NESTING_DEPTH; i++) {
      session()
          .execute(
              String.format(
                  "CREATE TYPE depth_%s (value frozen<depth_%s>)",
                  String.valueOf(i), String.valueOf(i - 1)));
    }

    session()
        .execute(
            String.format(
                "CREATE TABLE nested_udt_table (a int PRIMARY KEY, b frozen<depth_0>, c frozen<depth_1>, d frozen<depth_2>, e frozen<depth_3>,"
                    + "f frozen<depth_%s>)",
                MAX_NESTING_DEPTH));

    // insert UDT data
    KeyspaceMetadata keyspaceMetadata = cluster().getMetadata().getKeyspace(keyspace);
    UserType depthZeroDef = keyspaceMetadata.getUserType("depth_0");
    UDTValue depthZero = depthZeroDef.newValue().setInt("age", 42).setString("name", "Bob");

    UserType depthOneDef = keyspaceMetadata.getUserType("depth_1");
    UDTValue depthOne = depthOneDef.newValue().setUDTValue("value", depthZero);

    UserType depthTwoDef = keyspaceMetadata.getUserType("depth_2");
    UDTValue depthTwo = depthTwoDef.newValue().setUDTValue("value", depthOne);

    UserType depthThreeDef = keyspaceMetadata.getUserType("depth_3");
    UDTValue depthThree = depthThreeDef.newValue().setUDTValue("value", depthTwo);

    UserType depthFourDef = keyspaceMetadata.getUserType("depth_4");
    UDTValue depthFour = depthFourDef.newValue().setUDTValue("value", depthThree);

    PreparedStatement ins =
        session()
            .prepare("INSERT INTO nested_udt_table (a, b, c, d, e, f) VALUES (?, ?, ?, ?, ?, ?)");
    session().execute(ins.bind(0, depthZero, depthOne, depthTwo, depthThree, depthFour));

    // retrieve and verify data
    ResultSet rs = session().execute("SELECT * FROM nested_udt_table");
    List<Row> rows = rs.all();
    assertThat(rows.size()).isEqualTo(1);

    Row row = rows.get(0);

    assertThat(row.getInt("a")).isEqualTo(0);
    assertThat(row.getUDTValue("b")).isEqualTo(depthZero);
    assertThat(row.getUDTValue("c")).isEqualTo(depthOne);
    assertThat(row.getUDTValue("d")).isEqualTo(depthTwo);
    assertThat(row.getUDTValue("e")).isEqualTo(depthThree);
    assertThat(row.getUDTValue("f")).isEqualTo(depthFour);
  }

  @Test(groups = "short")
  public void should_save_and_retrieve_UDTs_with_null_values() throws Exception {
    // create UDT
    session().execute("CREATE TYPE user_null_values (a text, b int, c uuid, d blob)");
    session()
        .execute("CREATE TABLE null_values_table (a int PRIMARY KEY, b frozen<user_null_values>)");

    // insert UDT data
    UserType userTypeDef =
        cluster().getMetadata().getKeyspace(keyspace).getUserType("user_null_values");
    UDTValue userType =
        userTypeDef
            .newValue()
            .setString("a", null)
            .setInt("b", 0)
            .setUUID("c", null)
            .setBytes("d", null);

    PreparedStatement ins = session().prepare("INSERT INTO null_values_table (a, b) VALUES (?, ?)");
    session().execute(ins.bind(0, userType));

    // retrieve and verify data
    ResultSet rs = session().execute("SELECT * FROM null_values_table");
    List<Row> rows = rs.all();
    assertThat(rows.size()).isEqualTo(1);

    Row row = rows.get(0);

    assertThat(row.getInt("a")).isEqualTo(0);
    assertThat(row.getUDTValue("b")).isEqualTo(userType);

    // test empty strings
    userType =
        userTypeDef
            .newValue()
            .setString("a", "")
            .setInt("b", 0)
            .setUUID("c", null)
            .setBytes("d", ByteBuffer.allocate(0));
    session().execute(ins.bind(0, userType));

    // retrieve and verify data
    rs = session().execute("SELECT * FROM null_values_table");
    rows = rs.all();
    assertThat(rows.size()).isEqualTo(1);

    row = rows.get(0);

    assertThat(row.getInt("a")).isEqualTo(0);
    assertThat(row.getUDTValue("b")).isEqualTo(userType);
  }

  @Test(groups = "short")
  public void should_save_and_retrieve_UDTs_with_null_collections() throws Exception {
    // create UDT
    session()
        .execute(
            "CREATE TYPE user_null_collections (a List<text>, b Set<text>, c Map<text, text>, d frozen<Tuple<text>>)");
    session()
        .execute(
            "CREATE TABLE null_collections_table (a int PRIMARY KEY, b frozen<user_null_collections>)");

    // insert null UDT data
    PreparedStatement ins =
        session()
            .prepare(
                "INSERT INTO null_collections_table (a, b) "
                    + "VALUES (0, { a: ?, b: ?, c: ?, d: ? })");
    session()
        .execute(
            ins.bind().setList(0, null).setSet(1, null).setMap(2, null).setTupleValue(3, null));

    // retrieve and verify data
    ResultSet rs = session().execute("SELECT * FROM null_collections_table");
    List<Row> rows = rs.all();
    assertThat(rows.size()).isEqualTo(1);

    Row row = rows.get(0);
    assertThat(row.getInt("a")).isEqualTo(0);

    UserType userTypeDef =
        cluster().getMetadata().getKeyspace(keyspace).getUserType("user_null_collections");
    UDTValue userType =
        userTypeDef
            .newValue()
            .setList("a", null)
            .setSet("b", null)
            .setMap("c", null)
            .setTupleValue("d", null);
    assertThat(row.getUDTValue("b")).isEqualTo(userType);

    // test missing UDT args
    ins = session().prepare("INSERT INTO null_collections_table (a, b) " + "VALUES (1, { a: ? })");
    session().execute(ins.bind().setList(0, new ArrayList<Object>()));

    // retrieve and verify data
    rs = session().execute("SELECT * FROM null_collections_table");
    rows = rs.all();
    assertThat(rows.size()).isEqualTo(2);

    row = rows.get(0);
    assertThat(row.getInt("a")).isEqualTo(1);

    userType = userTypeDef.newValue().setList(0, new ArrayList<Object>());
    assertThat(row.getUDTValue("b")).isEqualTo(userType);
  }

  @Test(groups = "short")
  public void should_indicate_user_type_is_frozen() {
    session()
        .execute("CREATE TABLE frozen_table(k int primary key, v frozen<type_for_frozen_test>)");

    KeyspaceMetadata keyspaceMetadata = cluster().getMetadata().getKeyspace(this.keyspace);

    assertThat(keyspaceMetadata.getUserType("type_for_frozen_test")).isNotFrozen();

    DataType userType = keyspaceMetadata.getTable("frozen_table").getColumn("v").getType();
    assertThat(userType).isFrozen();
    assertThat(userType.toString()).isEqualTo("frozen<" + keyspace + ".type_for_frozen_test>");

    // The frozen flag is not set for result set definitions (the protocol does not provide
    // that information and it's not really useful in that situation). We always return false.
    ResultSet rs = session().execute("SELECT v FROM frozen_table WHERE k = 1");
    assertThat(rs.getColumnDefinitions().getType(0)).isNotFrozen();

    // Same thing for prepared statements
    PreparedStatement pst = session().prepare("SELECT v FROM frozen_table WHERE k = ?");
    assertThat(pst.getVariables().getType(0)).isNotFrozen();
  }

  @Test(groups = "short")
  @CassandraVersion(value = "3.6", description = "Non-frozen UDTs were introduced in C* 3.6")
  public void should_indicate_user_type_is_not_frozen() {
    session().execute("CREATE TABLE not_frozen_table(k int primary key, v type_for_frozen_test)");

    KeyspaceMetadata keyspaceMetadata = cluster().getMetadata().getKeyspace(this.keyspace);

    assertThat(keyspaceMetadata.getUserType("type_for_frozen_test")).isNotFrozen();

    DataType userType = keyspaceMetadata.getTable("not_frozen_table").getColumn("v").getType();
    assertThat(userType).isNotFrozen();
    assertThat(userType.toString()).isEqualTo(keyspace + ".type_for_frozen_test");

    ResultSet rs = session().execute("SELECT v FROM not_frozen_table WHERE k = 1");
    assertThat(rs.getColumnDefinitions().getType(0)).isNotFrozen();

    PreparedStatement pst = session().prepare("SELECT v FROM not_frozen_table WHERE k = ?");
    assertThat(pst.getVariables().getType(0)).isNotFrozen();
  }

  @Test(groups = "short")
  public void should_handle_udt_named_like_a_collection() {
    execute(
        "CREATE TYPE tuple(a text)",
        "CREATE TYPE list(a text)",
        "CREATE TYPE frozen(a text)",
        "CREATE TYPE udt(tuple frozen<tuple>, frozen frozen<frozen>, "
            + "m map<frozen<list>,frozen<frozen>>)");
    UserType udt = cluster().getMetadata().getKeyspace(keyspace).getUserType("udt");
    assertThat(udt.getFieldType("tuple")).isInstanceOf(UserType.class);
    assertThat(udt.getFieldType("frozen")).isInstanceOf(UserType.class);
    assertThat((udt.getFieldType("m").getTypeArguments().get(0))).isInstanceOf(UserType.class);
    assertThat((udt.getFieldType("m").getTypeArguments().get(1))).isInstanceOf(UserType.class);
  }
}
