src/java/org/apache/cassandra/db/context/CounterContext.java
/**
     * Return a context w/ an aggregated count for each node id.
     *
     * @param left counter context.
     * @param right counter context.
     * @param allocator An allocator for the merged value.
     */
Return a context w/ an aggregated count for each node id.
@param left counter context.
@param right counter context.
@param allocator An allocator for the merged value.
Return a context w / an aggregated count for each node id . @param left counter context . @param right counter context . @param allocator An allocator for the merged value .
@Override
    public ByteBuffer merge(ByteBuffer left, ByteBuffer right, Allocator allocator)
    {
        ContextState leftState = new ContextState(left, headerLength(left));
        ContextState rightState = new ContextState(right, headerLength(right));
        int mergedHeaderLength = HEADER_SIZE_LENGTH;
        int mergedBodyLength = 0;
        while (leftState.hasRemaining() && rightState.hasRemaining())
        {
            int cmp = leftState.compareIdTo(rightState);
            if (cmp == 0)
            {
                mergedBodyLength += STEP_LENGTH;
                if (leftState.isDelta() || rightState.isDelta())
                    mergedHeaderLength += HEADER_ELT_LENGTH;
                leftState.moveToNext();
                rightState.moveToNext();
            }
            else if (cmp > 0)
            {
                mergedBodyLength += STEP_LENGTH;
                if (rightState.isDelta())
                    mergedHeaderLength += HEADER_ELT_LENGTH;
                rightState.moveToNext();
            }
            else 
            {
                mergedBodyLength += STEP_LENGTH;
                if (leftState.isDelta())
                    mergedHeaderLength += HEADER_ELT_LENGTH;
                leftState.moveToNext();
            }
        }
        mergedHeaderLength += leftState.remainingHeaderLength() + rightState.remainingHeaderLength();
        mergedBodyLength += leftState.remainingBodyLength() + rightState.remainingBodyLength();
        ByteBuffer merged = allocator.allocate(mergedHeaderLength + mergedBodyLength);
        merged.putShort(merged.position(), (short) ((mergedHeaderLength - HEADER_SIZE_LENGTH) / HEADER_ELT_LENGTH));
        ContextState mergedState = new ContextState(merged, mergedHeaderLength);
        leftState.reset();
        rightState.reset();
        while (leftState.hasRemaining() && rightState.hasRemaining())
        {
            int cmp = leftState.compareIdTo(rightState);
            if (cmp == 0)
            {
                if (leftState.isDelta() || rightState.isDelta())
                {
                    if (leftState.isDelta() && rightState.isDelta())
                    {
                        long clock = leftState.getClock() + rightState.getClock();
                        long count = leftState.getCount() + rightState.getCount();
                        if (leftState.getTimestamp() > rightState.getTimestamp()) {
                            mergedState.writeElement(leftState.getNodeId(), clock, count, leftState.getTimestamp(), leftState.getCoordinatorKey(), true);
                        } else {
                            mergedState.writeElement(rightState.getNodeId(), clock, count, rightState.getTimestamp(), rightState.getCoordinatorKey(), true);
                        }
                    }
                    else
                    {
                        (leftState.isDelta() ? leftState : rightState).copyTo(mergedState);
                    }
                }
                else
                {
                    long leftClock = leftState.getClock();
                    long rightClock = rightState.getClock();
                    if (leftClock == rightClock)
                    {
                        long leftCount = leftState.getCount();
                        long rightCount = rightState.getCount();
                        if (leftCount != rightCount)
                        {
                            logger.error("invalid counter shard detected; ({}, {}, {}) and ({}, {}, {}) differ only in "
                                    + "count; will pick highest to self-heal; this indicates a bug or corruption generated a bad counter shard",
                                    new Object[] {
                                            leftState.getNodeId(), leftClock, leftCount,
                                            rightState.getNodeId(), rightClock, rightCount,
                                     });
                        }
                        if (leftCount > rightCount)
                        {
                            leftState.copyTo(mergedState);
                        }
                        else
                        {
                            rightState.copyTo(mergedState);
                        }
                    }
                    else
                    {
                        if ((leftClock >= 0 && rightClock > 0 && leftClock >= rightClock)
                                || (leftClock < 0 && (rightClock > 0 || leftClock < rightClock)))
                            leftState.copyTo(mergedState);
                        else
                            rightState.copyTo(mergedState);
                    }
                }
                rightState.moveToNext();
                leftState.moveToNext();
            }
            else if (cmp > 0)
            {
                rightState.copyTo(mergedState);
                rightState.moveToNext();
            }
            else 
            {
                leftState.copyTo(mergedState);
                leftState.moveToNext();
            }
        }
        while (leftState.hasRemaining())
        {
            leftState.copyTo(mergedState);
            leftState.moveToNext();
        }
        while (rightState.hasRemaining())
        {
            rightState.copyTo(mergedState);
            rightState.moveToNext();
        }
        return merged;
    }
@ Override public ByteBuffer merge ( ByteBuffer left , ByteBuffer right , Allocator allocator ) { ContextState leftState = new ContextState ( left , headerLength ( left ) ) ; ContextState rightState = new ContextState ( right , headerLength ( right ) ) ; int mergedHeaderLength = HEADER_SIZE_LENGTH ; int mergedBodyLength = 0 ; while ( leftState . hasRemaining ( ) && rightState . hasRemaining ( ) ) { int cmp = leftState . compareIdTo ( rightState ) ; if ( cmp == 0 ) { mergedBodyLength += STEP_LENGTH ; if ( leftState . isDelta ( ) || rightState . isDelta ( ) ) mergedHeaderLength += HEADER_ELT_LENGTH ; leftState . moveToNext ( ) ; rightState . moveToNext ( ) ; } else if ( cmp > 0 ) { mergedBodyLength += STEP_LENGTH ; if ( rightState . isDelta ( ) ) mergedHeaderLength += HEADER_ELT_LENGTH ; rightState . moveToNext ( ) ; } else { mergedBodyLength += STEP_LENGTH ; if ( leftState . isDelta ( ) ) mergedHeaderLength += HEADER_ELT_LENGTH ; leftState . moveToNext ( ) ; } } mergedHeaderLength += leftState . remainingHeaderLength ( ) + rightState . remainingHeaderLength ( ) ; mergedBodyLength += leftState . remainingBodyLength ( ) + rightState . remainingBodyLength ( ) ; ByteBuffer merged = allocator . allocate ( mergedHeaderLength + mergedBodyLength ) ; merged . putShort ( merged . position ( ) , ( short ) ( ( mergedHeaderLength - HEADER_SIZE_LENGTH ) / HEADER_ELT_LENGTH ) ) ; ContextState mergedState = new ContextState ( merged , mergedHeaderLength ) ; leftState . reset ( ) ; rightState . reset ( ) ; while ( leftState . hasRemaining ( ) && rightState . hasRemaining ( ) ) { int cmp = leftState . compareIdTo ( rightState ) ; if ( cmp == 0 ) { if ( leftState . isDelta ( ) || rightState . isDelta ( ) ) { if ( leftState . isDelta ( ) && rightState . isDelta ( ) ) { long clock = leftState . getClock ( ) + rightState . getClock ( ) ; long count = leftState . getCount ( ) + rightState . getCount ( ) ; if ( leftState . getTimestamp ( ) > rightState . getTimestamp ( ) ) { mergedState . writeElement ( leftState . getNodeId ( ) , clock , count , leftState . getTimestamp ( ) , leftState . getCoordinatorKey ( ) , true ) ; } else { mergedState . writeElement ( rightState . getNodeId ( ) , clock , count , rightState . getTimestamp ( ) , rightState . getCoordinatorKey ( ) , true ) ; } } else { ( leftState . isDelta ( ) ? leftState : rightState ) . copyTo ( mergedState ) ; } } else { long leftClock = leftState . getClock ( ) ; long rightClock = rightState . getClock ( ) ; if ( leftClock == rightClock ) { long leftCount = leftState . getCount ( ) ; long rightCount = rightState . getCount ( ) ; if ( leftCount != rightCount ) { logger . error ( "invalid counter shard detected; ({}, {}, {}) and ({}, {}, {}) differ only in " + "count; will pick highest to self-heal; this indicates a bug or corruption generated a bad counter shard" , new Object [ ] { leftState . getNodeId ( ) , leftClock , leftCount , rightState . getNodeId ( ) , rightClock , rightCount , } ) ; } if ( leftCount > rightCount ) { leftState . copyTo ( mergedState ) ; } else { rightState . copyTo ( mergedState ) ; } } else { if ( ( leftClock >= 0 && rightClock > 0 && leftClock >= rightClock ) || ( leftClock < 0 && ( rightClock > 0 || leftClock < rightClock ) ) ) leftState . copyTo ( mergedState ) ; else rightState . copyTo ( mergedState ) ; } } rightState . moveToNext ( ) ; leftState . moveToNext ( ) ; } else if ( cmp > 0 ) { rightState . copyTo ( mergedState ) ; rightState . moveToNext ( ) ; } else { leftState . copyTo ( mergedState ) ; leftState . moveToNext ( ) ; } } while ( leftState . hasRemaining ( ) ) { leftState . copyTo ( mergedState ) ; leftState . moveToNext ( ) ; } while ( rightState . hasRemaining ( ) ) { rightState . copyTo ( mergedState ) ; rightState . moveToNext ( ) ; } return merged ; }
Return a context w/ an aggregated count for each node id.
Return a context w / an aggregated count for each node id .
// Compute size of result // cmp < 0 // Do the actual merge // Local id and at least one is a delta // both delta, sum //WL TODO: Is this correct? // Only one have delta, keep that one // We should never see non-delta shards w/ same id+clock but different counts. However, if we do // we should "heal" the problem by being deterministic in our selection of shard - and // log the occurrence so that the operator will know something is wrong. // cmp < 0