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;
  }
@ 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.
Convert a Record into a fully - bound statement .