cassandra-protolib/src/main/java/com/streamsets/pipeline/stage/destination/cassandra/CassandraTarget.java
@SuppressWarnings("unchecked")
  private BoundStatement recordToBoundStatement(Record record) throws StageException {
    ImmutableList.Builder<Object> values = new ImmutableList.Builder<>();
    SortedSet<String> columnsPresent = Sets.newTreeSet(columnMappings.keySet());
    for (Map.Entry<String, String> mapping : columnMappings.entrySet()) {
      String columnName = mapping.getKey();
      String fieldPath = mapping.getValue();

      // If we're missing fields, skip them.
      // If a field is present, but null, also remove it from columnsPresent since we can't write nulls.
      if (!record.has(fieldPath) || record.get(fieldPath).getValue() == null) {
        columnsPresent.remove(columnName);
        continue;
      }

      final Object value = record.get(fieldPath).getValue();
      // Special cases for handling SDC Lists and Maps,
      // basically unpacking them into raw types.
      if (value instanceof List) {
        List<Object> unpackedList = new ArrayList<>();
        for (Field item : (List<Field>) value) {
          unpackedList.add(item.getValue());
        }
        values.add(unpackedList);
      } else if (value instanceof Map) {
        Map<Object, Object> unpackedMap = new HashMap<>();
        for (Map.Entry<String, Field> entry : ((Map<String, Field>) value).entrySet()) {
          unpackedMap.put(entry.getKey(), entry.getValue().getValue());
        }
        values.add(unpackedMap);
      } else {
        values.add(value);
      }
    }


    PreparedStatement stmt = statementCache.getUnchecked(columnsPresent);
    // .toArray required to pass in a list to a varargs method.
    Object[] valuesArray = values.build().toArray();
    BoundStatement boundStmt = null;
    try {
      boundStmt = stmt.bind(valuesArray);
    } catch (CodecNotFoundException | InvalidTypeException | NullPointerException e) {
      // NPE can occur if one of the values is a collection type with a null value inside it. Thus, it's a record
      // error. Note that this runs the risk of mistakenly treating a bug as a record error.
      // CodecNotFound is caused when there is no type conversion definition available from the provided type
      // to the target type.
      errorRecordHandler.onError(
          new OnRecordErrorException(
              record,
              Errors.CASSANDRA_06,
              record.getHeader().getSourceId(),
              e.toString(),
              e
          )
      );
    }
    return boundStmt;
  }
@SuppressWarnings("unchecked")
  private BoundStatement recordToBoundStatement(Record record) throws StageException {
    ImmutableList.Builder<Object> values = new ImmutableList.Builder<>();
    SortedSet<String> columnsPresent = Sets.newTreeSet(columnMappings.keySet());
    for (Map.Entry<String, String> mapping : columnMappings.entrySet()) {
      String columnName = mapping.getKey();
      String fieldPath = mapping.getValue();

      // If we're missing fields, skip them.
      // If a field is present, but null, also remove it from columnsPresent since we can't write nulls.
      if (!record.has(fieldPath) || record.get(fieldPath).getValue() == null) {
        columnsPresent.remove(columnName);
        continue;
      }

      final Object value = record.get(fieldPath).getValue();
      // Special cases for handling SDC Lists and Maps,
      // basically unpacking them into raw types.
      if (value instanceof List) {
        List<Object> unpackedList = new ArrayList<>();
        for (Field item : (List<Field>) value) {
          unpackedList.add(item.getValue());
        }
        values.add(unpackedList);
      } else if (value instanceof Map) {
        Map<Object, Object> unpackedMap = new HashMap<>();
        for (Map.Entry<String, Field> entry : ((Map<String, Field>) value).entrySet()) {
          unpackedMap.put(entry.getKey(), entry.getValue().getValue());
        }
        values.add(unpackedMap);
      } else {
        values.add(value);
      }
    }


    PreparedStatement stmt = statementCache.getUnchecked(columnsPresent);
    // .toArray required to pass in a list to a varargs method.
    Object[] valuesArray = values.build().toArray();
    BoundStatement boundStmt = null;
    try {
      boundStmt = stmt.bind(valuesArray);
    } catch (CodecNotFoundException | InvalidTypeException | NullPointerException e) {
      // NPE can occur if one of the values is a collection type with a null value inside it. Thus, it's a record
      // error. Note that this runs the risk of mistakenly treating a bug as a record error.
      // CodecNotFound is caused when there is no type conversion definition available from the provided type
      // to the target type.
      errorRecordHandler.onError(
          new OnRecordErrorException(
              record,
              Errors.CASSANDRA_06,
              record.getHeader().getSourceId(),
              e.toString(),
              e
          )
      );
    }
    return boundStmt;
  }
Convert a Record into a fully-bound statement.