tools/stress/src/org/apache/cassandra/stress/operations/DynamicWorkload.java
//This is a copy of MultiGetter.run with columnsPerKey and keysPerRead being used instead of the session parameters
This is a copy of MultiGetter.run with columnsPerKey and keysPerRead being used instead of the session parameters
This is a copy of MultiGetter . run with columnsPerKey and keysPerRead being used instead of the session parameters
public void read(ClientLibrary clientLibrary, int columnsPerKey, int keysPerRead) throws IOException
    {
        SlicePredicate nColumnsPredicate = new SlicePredicate().setSlice_range(new SliceRange(ByteBufferUtil.EMPTY_BYTE_BUFFER,
                                                                                      ByteBufferUtil.EMPTY_BYTE_BUFFER,
                                                                                      false, columnsPerKey));
        int offset = index * session.getKeysPerThread();
        Map<ByteBuffer,List<ColumnOrSuperColumn>> results;
        int columnCount = 0;
        long bytesCount = 0;
        if (session.getColumnFamilyType() == ColumnFamilyType.Super)
        {
            List<ByteBuffer> keys = generateKeys(offset, offset + keysPerRead);
            for (int j = 0; j < session.getSuperColumns(); j++)
            {
                ColumnParent parent = new ColumnParent("Super1").setSuper_column(ByteBufferUtil.bytes("S" + j));
                long startNano = System.nanoTime();
                boolean success = false;
                String exceptionMessage = null;
                for (int t = 0; t < session.getRetryTimes(); t++)
                {
                    if (success)
                        break;
                    try
                    {
			columnCount = 0;
                        results = clientLibrary.transactional_multiget_slice(keys, parent, nColumnsPredicate);
                        success = (results.size() > 0);
                        if (!success)
                            exceptionMessage = "Wrong number of columns: " + results.size() + " instead of " + columnsPerKey*keysPerRead;
                        for (List<ColumnOrSuperColumn> result : results.values()) {
                            columnCount += result.size();
                            for (ColumnOrSuperColumn cosc : result) {
                                bytesCount += ColumnOrSuperColumnHelper.findLength(cosc);
                            }
                        }
                    }
                    catch (Exception e)
                    {
                        exceptionMessage = getExceptionMessage(e);
                    }
                }
                if (!success)
                {
                    error(String.format("Operation [%d] retried %d times - error on calling multiget_slice for keys %s %s%n",
                                        index,
                                        session.getRetryTimes(),
                                        keys,
                                        (exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
                }
                session.operations.getAndIncrement();
                session.keys.getAndAdd(keys.size());
                session.columnCount.getAndAdd(columnCount);
                session.bytes.getAndAdd(bytesCount);
                long latencyNano = System.nanoTime() - startNano;
                session.latency.getAndAdd(latencyNano/1000000);
                session.latencies.add(latencyNano/1000);
                offset += keysPerRead;
            }
        }
        else
        {
            ColumnParent parent = new ColumnParent("Standard1");
            List<ByteBuffer> keys = generateKeys(offset, offset + keysPerRead);
            long startNano = System.nanoTime();
            boolean success = false;
            String exceptionMessage = null;
            for (int t = 0; t < session.getRetryTimes(); t++)
            {
                if (success)
                    break;
                try
                {
		    columnCount = 0;
                    results = clientLibrary.transactional_multiget_slice(keys, parent, nColumnsPredicate);
                    for (List<ColumnOrSuperColumn> result : results.values()) {
                        columnCount += result.size();
                        for (ColumnOrSuperColumn cosc : result) {
                            bytesCount += ColumnOrSuperColumnHelper.findLength(cosc);
                        }
                    }
                    success = (columnCount == columnsPerKey*keysPerRead);
		    if (!success) {
			exceptionMessage = "Wrong number of columns: " + columnCount + " instead of " + columnsPerKey*keysPerRead + ": ";
			for (Entry<ByteBuffer, List<ColumnOrSuperColumn>> entry : results.entrySet()) {
			    String key = ByteBufferUtil.string(entry.getKey());
			    int colCount = entry.getValue().size();
			    exceptionMessage += colCount + " for " + key + " ";
			}
		    }
                }
                catch (Exception e)
                {
                    exceptionMessage = getExceptionMessage(e);
                    success = false;
                }
            }
            if (!success)
            {
                List<String> raw_keys = new ArrayList<String>();
                for (ByteBuffer key : keys) {
                    raw_keys.add(ByteBufferUtil.string(key));
                }
                error(String.format("Operation [%d] retried %d times - error on calling multiget_slice for keys %s %s%n",
                                    index,
                                    session.getRetryTimes(),
                                    raw_keys,
                                    (exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
            }
            session.operations.getAndIncrement();
            session.keys.getAndAdd(keys.size());
            session.columnCount.getAndAdd(columnCount);
            session.bytes.getAndAdd(bytesCount);
            long latencyNano = System.nanoTime() - startNano;
            session.latency.getAndAdd(latencyNano/1000000);
            session.latencies.add(latencyNano/1000);
            offset += keysPerRead;
        }
    }
public void read ( ClientLibrary clientLibrary , int columnsPerKey , int keysPerRead ) throws IOException { SlicePredicate nColumnsPredicate = new SlicePredicate ( ) . setSlice_range ( new SliceRange ( ByteBufferUtil . EMPTY_BYTE_BUFFER , ByteBufferUtil . EMPTY_BYTE_BUFFER , false , columnsPerKey ) ) ; int offset = index * session . getKeysPerThread ( ) ; Map < ByteBuffer , List < ColumnOrSuperColumn > > results ; int columnCount = 0 ; long bytesCount = 0 ; if ( session . getColumnFamilyType ( ) == ColumnFamilyType . Super ) { List < ByteBuffer > keys = generateKeys ( offset , offset + keysPerRead ) ; for ( int j = 0 ; j < session . getSuperColumns ( ) ; j ++ ) { ColumnParent parent = new ColumnParent ( "Super1" ) . setSuper_column ( ByteBufferUtil . bytes ( "S" + j ) ) ; long startNano = System . nanoTime ( ) ; boolean success = false ; String exceptionMessage = null ; for ( int t = 0 ; t < session . getRetryTimes ( ) ; t ++ ) { if ( success ) break ; try { columnCount = 0 ; results = clientLibrary . transactional_multiget_slice ( keys , parent , nColumnsPredicate ) ; success = ( results . size ( ) > 0 ) ; if ( ! success ) exceptionMessage = "Wrong number of columns: " + results . size ( ) + " instead of " + columnsPerKey * keysPerRead ; for ( List < ColumnOrSuperColumn > result : results . values ( ) ) { columnCount += result . size ( ) ; for ( ColumnOrSuperColumn cosc : result ) { bytesCount += ColumnOrSuperColumnHelper . findLength ( cosc ) ; } } } catch ( Exception e ) { exceptionMessage = getExceptionMessage ( e ) ; } } if ( ! success ) { error ( String . format ( "Operation [%d] retried %d times - error on calling multiget_slice for keys %s %s%n" , index , session . getRetryTimes ( ) , keys , ( exceptionMessage == null ) ? "" : "(" + exceptionMessage + ")" ) ) ; } session . operations . getAndIncrement ( ) ; session . keys . getAndAdd ( keys . size ( ) ) ; session . columnCount . getAndAdd ( columnCount ) ; session . bytes . getAndAdd ( bytesCount ) ; long latencyNano = System . nanoTime ( ) - startNano ; session . latency . getAndAdd ( latencyNano / 1000000 ) ; session . latencies . add ( latencyNano / 1000 ) ; offset += keysPerRead ; } } else { ColumnParent parent = new ColumnParent ( "Standard1" ) ; List < ByteBuffer > keys = generateKeys ( offset , offset + keysPerRead ) ; long startNano = System . nanoTime ( ) ; boolean success = false ; String exceptionMessage = null ; for ( int t = 0 ; t < session . getRetryTimes ( ) ; t ++ ) { if ( success ) break ; try { columnCount = 0 ; results = clientLibrary . transactional_multiget_slice ( keys , parent , nColumnsPredicate ) ; for ( List < ColumnOrSuperColumn > result : results . values ( ) ) { columnCount += result . size ( ) ; for ( ColumnOrSuperColumn cosc : result ) { bytesCount += ColumnOrSuperColumnHelper . findLength ( cosc ) ; } } success = ( columnCount == columnsPerKey * keysPerRead ) ; if ( ! success ) { exceptionMessage = "Wrong number of columns: " + columnCount + " instead of " + columnsPerKey * keysPerRead + ": " ; for ( Entry < ByteBuffer , List < ColumnOrSuperColumn > > entry : results . entrySet ( ) ) { String key = ByteBufferUtil . string ( entry . getKey ( ) ) ; int colCount = entry . getValue ( ) . size ( ) ; exceptionMessage += colCount + " for " + key + " " ; } } } catch ( Exception e ) { exceptionMessage = getExceptionMessage ( e ) ; success = false ; } } if ( ! success ) { List < String > raw_keys = new ArrayList < String > ( ) ; for ( ByteBuffer key : keys ) { raw_keys . add ( ByteBufferUtil . string ( key ) ) ; } error ( String . format ( "Operation [%d] retried %d times - error on calling multiget_slice for keys %s %s%n" , index , session . getRetryTimes ( ) , raw_keys , ( exceptionMessage == null ) ? "" : "(" + exceptionMessage + ")" ) ) ; } session . operations . getAndIncrement ( ) ; session . keys . getAndAdd ( keys . size ( ) ) ; session . columnCount . getAndAdd ( columnCount ) ; session . bytes . getAndAdd ( bytesCount ) ; long latencyNano = System . nanoTime ( ) - startNano ; session . latency . getAndAdd ( latencyNano / 1000000 ) ; session . latencies . add ( latencyNano / 1000 ) ; offset += keysPerRead ; } }
This is a copy of MultiGetter.run with columnsPerKey and keysPerRead being used instead of the session parameters
This is a copy of MultiGetter . run with columnsPerKey and keysPerRead being used instead of the session parameters
//TODO: Perhaps we should randomize which columns we grab? //TODO Count subcolumns returned by super columns and use that to check for success