Java 类com.datastax.driver.core.querybuilder.Batch 实例源码

@Override
public void removeAccessTokenUsingRefreshToken(OAuth2RefreshToken refreshToken) {
  String tokenValue = refreshToken.getValue();
  // Lookup RefreshTokenToAccessToken table for locating access token
  RefreshTokenToAccessToken refreshTokenToAccessToken = refreshTokenToAccessTokenRepository.findOne(tokenValue);
  if (refreshTokenToAccessToken != null) {
    String accessTokenKey = refreshTokenToAccessToken.getAccessTokenKey();
    AccessToken accessToken = accessTokenRepository.findOne(accessTokenKey);
    String jsonOAuth2AccessToken = accessToken.getoAuth2AccessToken();
    OAuth2AccessToken oAuth2AccessToken = OAuthUtil.deserializeOAuth2AccessToken(jsonOAuth2AccessToken);
    // Delete access token from all related tables
    List<RegularStatement> statementList = prepareRemoveAccessTokenStatements(oAuth2AccessToken);
    // Delete from RefreshTokenToAccessToken table
    Delete refreshTokenToAccessTokenDelete = CassandraTemplate.createDeleteQuery(RefreshTokenToAccessToken.TABLE, refreshTokenToAccessToken, null, cassandraTemplate.getConverter());
    statementList.add(refreshTokenToAccessTokenDelete);
    Batch batch = QueryBuilder.batch(statementList.toArray(new RegularStatement[statementList.size()]));
    cassandraTemplate.execute(batch);
  }
}

/**
 * Tunes CQL statement execution options (consistency level, fetch option and etc.).
 *
 * @param statement Statement.
 * @return Modified statement.
 */
private Statement tuneStatementExecutionOptions(Statement statement) {
    String qry = "";
    if (statement instanceof BoundStatement) {
        qry = ((BoundStatement)statement).preparedStatement().getQueryString().trim().toLowerCase();
    }
    else if (statement instanceof PreparedStatement) {
        qry = ((PreparedStatement)statement).getQueryString().trim().toLowerCase();
    }
    boolean readStatement = qry.startsWith("select");
    boolean writeStatement = statement instanceof Batch || statement instanceof BatchStatement ||
        qry.startsWith("insert") || qry.startsWith("delete") || qry.startsWith("update");
    if (readStatement && readConsistency != null) {
        statement.setConsistencyLevel(readConsistency);
    }
    if (writeStatement && writeConsistency != null) {
        statement.setConsistencyLevel(writeConsistency);
    }
    if (fetchSize != null) {
        statement.setFetchSize(fetchSize);
    }
    return statement;
}

/**
 * Tunes CQL statement execution options (consistency level, fetch option and etc.).
 *
 * @param statement Statement.
 * @return Modified statement.
 */
private Statement tuneStatementExecutionOptions(Statement statement) {
    String qry = "";

    if (statement instanceof BoundStatement)
        qry = ((BoundStatement)statement).preparedStatement().getQueryString().trim().toLowerCase();
    else if (statement instanceof PreparedStatement)
        qry = ((PreparedStatement)statement).getQueryString().trim().toLowerCase();

    boolean readStatement = qry.startsWith("select");
    boolean writeStatement = statement instanceof Batch || statement instanceof BatchStatement ||
        qry.startsWith("insert") || qry.startsWith("delete") || qry.startsWith("update");

    if (readStatement && readConsistency != null)
        statement.setConsistencyLevel(readConsistency);

    if (writeStatement && writeConsistency != null)
        statement.setConsistencyLevel(writeConsistency);

    if (fetchSize != null)
        statement.setFetchSize(fetchSize);

    return statement;
}

private void runBatchInsert(List<Insert> insertRequest) {
    try {
        Batch batch;
        if (config.getLoggedBatch()) {
            batch = QueryBuilder.batch(insertRequest
                    .toArray(new RegularStatement[insertRequest.size()]));
        } else {
            batch = QueryBuilder.unloggedBatch(insertRequest
                    .toArray(new RegularStatement[insertRequest.size()]));
        }
        totalCassandraInsertRequest.addAndGet(insertRequest.size());
        ResultSetFuture future = cassandraSession.executeAsync(batch);
        CallBackListener listener = new CallBackListener(future, null);
        future.addListener(listener, pool);
        incrementBatchInsertCounter();
        pendingRequestCounter.incrementAndGet();
    } catch (Throwable ex) {
        LOGGER.error("Error publising metrics in MetricCassandraCollector:" + ex.getMessage());
        cassandraErrorCount.increment();
        registerError(ex);
    } finally {
        insertRequest.clear();
    }
}

private void runBatchUpdate(List<Update> updateRequest) {
    try {
        Batch batch;

        if (config.getLoggedBatch()) {
            batch = QueryBuilder.batch(updateRequest
                    .toArray(new RegularStatement[updateRequest.size()]));
        } else {
            batch = QueryBuilder.unloggedBatch(updateRequest
                    .toArray(new RegularStatement[updateRequest.size()]));
        }
        totalCassandraUpdateRequest.addAndGet(updateRequest.size());
        ResultSetFuture future = cassandraSession.executeAsync(batch);
        CallBackListener listener = new CallBackListener(future, null);
        future.addListener(listener, pool);
        incrementBatchUpdateCounter();
        pendingRequestCounter.incrementAndGet();
    } catch (Throwable ex) {
        LOGGER.error("Error publising metrics in MetricCassandraCollector:" + ex.getMessage());
        cassandraErrorCount.increment();
        registerError(ex);
    } finally {
        updateRequest.clear();
    }
}

public void deleteAcls(List<AclObjectIdentity> objectIdsToDelete) {
    assertAclObjectIdentityList(objectIdsToDelete);

    if (LOG.isDebugEnabled()) {
        LOG.debug("BEGIN deleteAcls: objectIdsToDelete: " + objectIdsToDelete);
    }

    List<String> ids = new ArrayList<String>();
    for (AclObjectIdentity entry : objectIdsToDelete) {
        ids.add(entry.getRowId());
    }
    Batch batch = QueryBuilder.batch();
    batch.add(QueryBuilder.delete().all().from(KEYSPACE, AOI_TABLE).where(QueryBuilder.in("id", ids.toArray())));
    batch.add(QueryBuilder.delete().all().from(KEYSPACE, CHILDREN_TABLE).where(QueryBuilder.in("id", ids.toArray())));
    session.execute(batch);

    if (LOG.isDebugEnabled()) {
        LOG.debug("END deleteAcls");
    }
}

public void saveAcl(AclObjectIdentity aoi) throws AclAlreadyExistsException {
    assertAclObjectIdentity(aoi);

    if (LOG.isDebugEnabled()) {
        LOG.debug("BEGIN saveAcl: aclObjectIdentity: " + aoi);
    }

    // Check this object identity hasn't already been persisted
    if (findAclObjectIdentity(aoi) != null) {
        throw new AclAlreadyExistsException("Object identity '" + aoi + "' already exists");
    }

    Batch batch = QueryBuilder.batch();
    batch.add(QueryBuilder.insertInto(KEYSPACE, AOI_TABLE).values(AOI_KEYS, new Object[] { aoi.getRowId(), aoi.getId(), aoi.getObjectClass(), aoi.isEntriesInheriting(),
            aoi.getOwnerId(), aoi.isOwnerPrincipal(), aoi.getParentObjectId(), aoi.getParentObjectClass() }));

    if (aoi.getParentRowId() != null) {
        batch.add(QueryBuilder.insertInto(KEYSPACE, CHILDREN_TABLE).values(CHILD_KEYS, new Object[] { aoi.getParentRowId(), aoi.getRowId(), aoi.getId(), aoi.getObjectClass() }));
    }
    session.execute(batch);

    if (LOG.isDebugEnabled()) {
        LOG.debug("END saveAcl");
    }
}

@Override
public void storeRefreshToken(OAuth2RefreshToken refreshToken, OAuth2Authentication authentication) {
  List<RegularStatement> statementList = new ArrayList<RegularStatement>();

  byte[] serializedRefreshToken = SerializationUtils.serialize(refreshToken);
  ByteBuffer bufferedRefreshToken = ByteBuffer.wrap(serializedRefreshToken);

  byte[] serializedAuthentication = SerializationUtils.serialize(authentication);
  ByteBuffer bufferedAuthentication = ByteBuffer.wrap(serializedAuthentication);

  WriteOptions refreshWriteOptions = new WriteOptions();
  if (refreshToken instanceof ExpiringOAuth2RefreshToken) {
    ExpiringOAuth2RefreshToken expiringRefreshToken = (ExpiringOAuth2RefreshToken) refreshToken;
    Date expiration = expiringRefreshToken.getExpiration();
    if (expiration != null) {
      int seconds = Long.valueOf((expiration.getTime() - System.currentTimeMillis()) / 1000L).intValue();
      refreshWriteOptions.setTtl(seconds);
    }
  }

  // Insert into RefreshToken table
  Insert accessInsert = CassandraTemplate.createInsertQuery(RefreshToken.TABLE, new RefreshToken(refreshToken.getValue(), bufferedRefreshToken), refreshWriteOptions, cassandraTemplate.getConverter());
  statementList.add(accessInsert);

  // Insert into RefreshTokenAuthentication table
  Insert authInsert = CassandraTemplate.createInsertQuery(RefreshTokenAuthentication.TABLE, new RefreshTokenAuthentication(refreshToken.getValue(), bufferedAuthentication), refreshWriteOptions, cassandraTemplate.getConverter());
  statementList.add(authInsert);

  Batch batch = QueryBuilder.batch(statementList.toArray(new RegularStatement[statementList.size()]));
  cassandraTemplate.execute(batch);
}

@Override
public void removeRefreshToken(OAuth2RefreshToken token) {
  String tokenValue = token.getValue();
  List<RegularStatement> statementList = new ArrayList<RegularStatement>();
  // Delete from RefreshToken table
  statementList.add(prepareDeleteByPrimaryKeyRegularStatement(RefreshToken.class, tokenValue));
  // Delete from RefreshTokenAuthentication table
  statementList.add(prepareDeleteByPrimaryKeyRegularStatement(RefreshTokenAuthentication.class, tokenValue));
  // Delete from RefreshTokenToAccessToken table
  statementList.add(prepareDeleteByPrimaryKeyRegularStatement(RefreshTokenToAccessToken.class, tokenValue));
  Batch batch = QueryBuilder.batch(statementList.toArray(new RegularStatement[statementList.size()]));
  cassandraTemplate.execute(batch);
}

@SafeVarargs
public final CassandraUtils insert(Map<String, String>... paramss) {

    Batch batch = QueryBuilder.unloggedBatch();
    for (Map<String, String> params : paramss) {
        String columns = "";
        String values = "";
        for (String s : params.keySet()) {
            if (!s.equals(indexColumn)) {
                columns += s + ",";
                values = values + params.get(s) + ",";
            }
        }
        columns = columns.substring(0, columns.length() - 1);
        values = values.substring(0, values.length() - 1);

        batch.add(new SimpleStatement(new StringBuilder().append("INSERT INTO ")
                                                         .append(qualifiedTable)
                                                         .append(" (")
                                                         .append(columns)
                                                         .append(") VALUES (")
                                                         .append(values)
                                                         .append(");")
                                                         .toString()));
    }
    execute(batch);
    return this;
}

public static <W> void doCql3SaveToCassandra(RDD<W> rdd, ICassandraDeepJobConfig<W> writeConfig,
                                             Function1<W, Tuple2<Cells, Cells>> transformer) {
    if (!writeConfig.getIsWriteConfig()) {
        throw new IllegalArgumentException("Provided configuration object is not suitable for writing");
    }
    Tuple2<Map<String, ByteBuffer>, Map<String, ByteBuffer>> tuple = new Tuple2<>(null, null);

    RDD<Tuple2<Cells, Cells>> mappedRDD = rdd.map(transformer,
            ClassTag$.MODULE$.<Tuple2<Cells, Cells>>apply(tuple.getClass()));

    ((CassandraDeepJobConfig) writeConfig).createOutputTableIfNeeded(mappedRDD.first());

    final int pageSize = writeConfig.getBatchSize();
    int offset = 0;

    List<Tuple2<Cells, Cells>> elements = Arrays.asList((Tuple2<Cells, Cells>[]) mappedRDD.collect());
    List<Tuple2<Cells, Cells>> split;
    do {
        split = elements.subList(pageSize * (offset++), Math.min(pageSize * offset, elements.size()));

        Batch batch = QueryBuilder.batch();

        for (Tuple2<Cells, Cells> t : split) {
            Tuple2<String[], Object[]> bindVars = Utils.prepareTuple4CqlDriver(t);

            Insert insert = QueryBuilder
                    .insertInto(quote(writeConfig.getKeyspace()), quote(writeConfig.getTable()))
                    .values(bindVars._1(), bindVars._2());

            batch.add(insert);
        }
        writeConfig.getSession().execute(batch);

    } while (!split.isEmpty() && split.size() == pageSize);
}

private Batch prepareBatch(RegularStatement... statement) {
  Batch batch;
  if (batchType != null && batchType.equals(CassandraBatchType.UNLOGGED)) {
    batch = QueryBuilder.unloggedBatch(statement);
  } else {
    batch = QueryBuilder.batch(statement);
  }
  batch.setConsistencyLevel(getWriteConsistencyLevel());
  return batch;
}

@Override
public <T> void delete(T obj, ConsistencyLevel consistency) {

    checkNotNull(obj, "object argument");
    checkNotNull(consistency, "consistency level argument");
    checkState(m_isOpen, format("%s is closed", getClass().getSimpleName()));

    Schema schema = getSchema(obj);
    Batch batchStatement = batch(QueryBuilder.delete().from(schema.getTableName())
            .where(eq(schema.getID().getName(), schema.getID().getValue(obj))));

    // Remove index entries
    for (ColumnSpec colSpec : schema.getColumns()) {
        if (colSpec.isIndexed()) {
            String tableName = indexTableName(schema.getTableName(), colSpec.getName());
            Clause columnClause = eq(colSpec.getName(), colSpec.getValue(obj));
            Clause idClause = eq(joinColumnName(schema.getTableName()), schema.getID().getValue(obj));

            batchStatement.add(QueryBuilder.delete().from(tableName).where(columnClause).and(idClause));
        }
    }

    // Remove one-to-many relationships
    for (OneToManySpec relSpec : schema.getOneToManys()) {
        String joinTable = joinTableName(schema.getTableName(), relSpec.getSchema().getTableName());
        batchStatement.add(
                QueryBuilder.delete().from(joinTable)
                    .where(eq(joinColumnName(schema.getTableName()), schema.getID().getValue(obj)))
        );
    }

    executeStatement(batchStatement, consistency);

    m_instanceCache.remove(getInstanceID(obj));

}

@Override
public void writeToPath(K rowKey,
                        Path path,
                        Object structuredValue,
                        BatchContext batchContext) {
    Batch batch = validateAndGetBatch(batchContext);

    validateArgs(rowKey, path);
    Object simplifiedStructure = writeMapper.convertValue(structuredValue, Object.class);
    Map<Path,Object> pathMap = Collections.singletonMap(path, simplifiedStructure);
    Map<Path,Object> objectMap = Decomposer.get().decompose(pathMap);

    batch = batchContext == null ? batch() : batch;
    List<Object> bindArguments = batchContext == null ?
                                    new ArrayList<Object>() :
                                    ((CqlBatchContext)batchContext).getBindArguments();
    Statement insertStatement = insertInto(tableName)
            .value(partitionKeyColumnName, bindMarker())
            .value(pathColumnName, bindMarker())
            .value(valueColumnName, bindMarker())
            .using(timestamp(getCurrentMicros()));
    insertStatement.setConsistencyLevel(defaultConsistencyLevel);


    for (Map.Entry<Path,Object> entry : objectMap.entrySet()) {
        batch.add(insertStatement);

        String stringValue = StructureConverter.get().toString(entry.getValue());

        bindArguments.add(rowKey);
        bindArguments.add(entry.getKey().toString());
        bindArguments.add(stringValue);
    }

    if (batchContext == null) {
        Query boundStatement = session.prepare(batch.getQueryString()).bind(bindArguments.toArray());
        boundStatement.setConsistencyLevel(defaultConsistencyLevel);
        session.execute(boundStatement);
    }
}

private Batch validateAndGetBatch(BatchContext batchContext) {
    if (batchContext == null) {
        return null;
    }
    if (!(batchContext instanceof CqlBatchContext)) {
        throw new IllegalArgumentException("batchContext is not a CQL batch context");
    }
    return ((CqlBatchContext)batchContext).getBatch();
}

@Override
public void removeAccessToken(OAuth2AccessToken token) {
  List<RegularStatement> statementList = prepareRemoveAccessTokenStatements(token);
  Batch batch = QueryBuilder.batch(statementList.toArray(new RegularStatement[statementList.size()]));
  cassandraTemplate.execute(batch);
}

@Override
public <T> T create(T object, ConsistencyLevel consistency) {

    checkNotNull(object, "object argument");
    checkNotNull(consistency, "consistency argument");
    checkState(m_isOpen, format("%s is closed", getClass().getSimpleName()));
    checkArgument(
            object.getClass().isAnnotationPresent(ENTITY),
            format("%s not annotated with @%s", getClass().getSimpleName(), ENTITY.getCanonicalName()));

    Schema schema = getSchema(object);

    checkArgument(
            schema.getID().getValue(object) == null,
            format("property annotated with @%s must be null", ID.getCanonicalName()));

    // Object persistence (incl. indices)
    UUID id = UUID.randomUUID();
    Batch batch = batch();
    Insert insertStatement = insertInto(schema.getTableName()).value(schema.getID().getName(), id);

    for (ColumnSpec colSpec : schema.getColumns()) {

        insertStatement.value(colSpec.getName(), colSpec.getValue(object));

        if (colSpec.isIndexed()) {
            String tableName = indexTableName(schema.getTableName(), colSpec.getName());
            batch.add(
                    insertInto(tableName)
                        .value(colSpec.getName(), colSpec.getValue(object))
                        .value(joinColumnName(schema.getTableName()), id)
            );
        }
    }

    batch.add(insertStatement);

    // One-to-Many relationship persistence
    for (OneToManySpec relationSpec : schema.getOneToManys()) {
        Schema s = relationSpec.getSchema();

        Object relations = relationSpec.getValue(object);

        if (relations == null) {
            continue;
        }

        for (Object item : (Collection<?>) relations) {
            UUID relationID = (UUID) s.getID().getValue(item);

            if (relationID == null) {
                throw new IllegalStateException(
                        "encountered relation with null ID property (entity not persisted?)");
            }

            String joinTable = joinTableName(schema.getTableName(), s.getTableName());

            batch.add(
                    insertInto(joinTable)
                        .value(joinColumnName(schema.getTableName()), id)
                        .value(joinColumnName(s.getTableName()), relationID)
            );
        }

    }

    executeStatement(batch, consistency);

    schema.getID().setValue(object, id);
    cacheInstance(object);

    return object;
}

@Override
public void deletePath(K rowKey, Path path, BatchContext batchContext) {
    Batch batch = validateAndGetBatch(batchContext);

    validateArgs(rowKey, path);

    // converting from a string and back normalizes the path, e.g. makes sure ends with the delimiter character
    String start = path.toString();
    String finish = getFinishString(start);

    // would like to just do a delete with a where clause, but unfortunately Cassandra can't do that in CQL (either)
    // with >= and <=

    // Since the path column is in the primary key, we need to just delete whole rows.

    Object[] args = {rowKey,start,finish};
    ResultSet resultSet = session.execute(readForDeleteQuery.bind(args));
    if (resultSet.isExhausted()) {
        // not found
        return;
    }

    Delete deleteStatement = delete().from(tableName);
    deleteStatement
            .using(timestamp(getCurrentMicros()))
            .where(eq(partitionKeyColumnName, rowKey))
            .and(eq(pathColumnName, bindMarker()));

    batch = batchContext == null ? batch() : batch;
    List<Object> bindArguments = batchContext == null ?
            new ArrayList<Object>() :
            ((CqlBatchContext)batchContext).getBindArguments();

    for (Row row : resultSet) {
        String pathToDelete = row.getString(0);
        batch.add(deleteStatement);
        bindArguments.add(pathToDelete);
    }

    if (batchContext == null) {
        BoundStatement query = session.prepare(batch.getQueryString()).bind(bindArguments.toArray());
        query.setConsistencyLevel(defaultConsistencyLevel);
        session.execute(query);
    }
}

private Batch getBatch() {
    return batch;
}

public void updateAcl(AclObjectIdentity aoi, List<AclEntry> entries) throws AclNotFoundException {
    assertAclObjectIdentity(aoi);

    if (LOG.isDebugEnabled()) {
        LOG.debug("BEGIN updateAcl: aclObjectIdentity: " + aoi + ", entries: " + entries);
    }

    // Check this object identity is already persisted
    AclObjectIdentity persistedAoi = findAclObjectIdentity(aoi);
    if (persistedAoi == null) {
        throw new AclNotFoundException("Object identity '" + aoi + "' does not exist");
    }

    // Update AOI & delete existing ACLs
    Batch batch = QueryBuilder.batch();
    batch.add(QueryBuilder.insertInto(KEYSPACE, AOI_TABLE).values(AOI_KEYS, new Object[] { aoi.getRowId(), aoi.getId(), aoi.getObjectClass(), aoi.isEntriesInheriting(),
            aoi.getOwnerId(), aoi.isOwnerPrincipal(), aoi.getParentObjectId(), aoi.getParentObjectClass() }));
    batch.add(QueryBuilder.delete().all().from(KEYSPACE, ACL_TABLE).where(QueryBuilder.eq("id", aoi.getRowId())));

    // Check if parent is different and delete from children table
    boolean parentChanged = false;
    if (!(persistedAoi.getParentRowId() == null ? aoi.getParentRowId() == null : persistedAoi.getParentRowId().equals(aoi.getParentRowId()))) {
        parentChanged = true;

        if (persistedAoi.getParentRowId() != null) {
            batch.add(QueryBuilder.delete().all().from(KEYSPACE, CHILDREN_TABLE).where(QueryBuilder.eq("id", persistedAoi.getParentRowId())).and(QueryBuilder.eq("childId", aoi.getRowId())));
        }           
    }
    session.execute(batch);

    // Update ACLs & children table 
    batch = QueryBuilder.batch();
    boolean executeBatch = false;

    if (entries != null && !entries.isEmpty()) {
        for (AclEntry entry : entries) {
            batch.add(QueryBuilder.insertInto(KEYSPACE, ACL_TABLE).values(ACL_KEYS, new Object[] { aoi.getRowId(), entry.getOrder(), entry.getSid(), entry.getMask(), entry.isSidPrincipal(),
                    entry.isGranting(), entry.isAuditSuccess(), entry.isAuditFailure() }));
        }
        executeBatch = true;
    }       
    if (parentChanged) {
        if (aoi.getParentRowId() != null) {
            batch.add(QueryBuilder.insertInto(KEYSPACE, CHILDREN_TABLE).values(CHILD_KEYS, new Object[] { aoi.getParentRowId(), aoi.getRowId(), aoi.getId(), aoi.getObjectClass() }));
        }
        executeBatch = true;
    }
    if (executeBatch) {
        session.execute(batch);
    }       

    if (LOG.isDebugEnabled()) {
        LOG.debug("END updateAcl");
    }
}

@Test
public void testInsertDbObjectsBatch() throws Exception {
    testInSession(new Callback() {
        @Override
        public void call(Session session) throws Exception {


            Batch bs = QueryBuilder.batch();

            bs.add(new SimpleStatement(QUERY));
            bs.add(new SimpleStatement(QUERY));


            PreparedStatement preparedStatement = session.prepare(bs);

            DatastaxBinder<List<DbObject>> datastaxBinder = DatastaxMapperFactory.newInstance().useAsm(false).mapFrom(new TypeReference<List<DbObject>>() {
            });

            Statement statement = datastaxBinder.mapTo(dbObjects, preparedStatement);

            statement.enableTracing();
            session.execute(statement);

            checkObjectInserted(session, 0);
            checkObjectInserted(session, 1);

        }
    });
}