diff --git a/src/org/apache/cassandra/db/Row.java b/src/org/apache/cassandra/db/Row.java index 1e429c3ba2..069b16a6f2 100644 --- a/src/org/apache/cassandra/db/Row.java +++ b/src/org/apache/cassandra/db/Row.java @@ -133,32 +133,30 @@ public class Row * and return the resultant row. This assumes that the row that * is being submitted is a super set of the current row so * it only calculates additional - * difference and does not take care of what needs to be delted from the current row to make + * difference and does not take care of what needs to be removed from the current row to make * it same as the input row. */ - public Row diff(Row row) + public Row diff(Row rowNew) { Row rowDiff = new Row(key_); - Map columnFamilies = row.getColumnFamilyMap(); - Set cfNames = columnFamilies.keySet(); - for (String cfName : cfNames) + for (ColumnFamily cfNew : rowNew.getColumnFamilies()) { - ColumnFamily cf = columnFamilies_.get(cfName); + ColumnFamily cf = columnFamilies_.get(cfNew.name()); ColumnFamily cfDiff = null; if (cf == null) - rowDiff.getColumnFamilyMap().put(cfName, columnFamilies.get(cfName)); + rowDiff.addColumnFamily(cfNew); else { - cfDiff = cf.diff(columnFamilies.get(cfName)); + cfDiff = cf.diff(cfNew); if (cfDiff != null) - rowDiff.getColumnFamilyMap().put(cfName, cfDiff); + rowDiff.addColumnFamily(cfDiff); } } - if (rowDiff.getColumnFamilyMap().size() != 0) - return rowDiff; - else + if (rowDiff.getColumnFamilies().isEmpty()) return null; + else + return rowDiff; } public Row cloneMe() diff --git a/src/org/apache/cassandra/db/SuperColumn.java b/src/org/apache/cassandra/db/SuperColumn.java index cd522d0a70..e1b0efad8f 100644 --- a/src/org/apache/cassandra/db/SuperColumn.java +++ b/src/org/apache/cassandra/db/SuperColumn.java @@ -232,10 +232,9 @@ public final class SuperColumn implements IColumn, Serializable public IColumn diff(IColumn column) { - IColumn columnDiff = new SuperColumn(column.name()); - Collection columns = column.getSubColumns(); + IColumn columnDiff = new SuperColumn(column.name()); - for ( IColumn subColumn : columns ) + for (IColumn subColumn : column.getSubColumns()) { IColumn columnInternal = columns_.get(subColumn.name()); if(columnInternal == null ) @@ -251,7 +250,8 @@ public final class SuperColumn implements IColumn, Serializable } } } - if(columnDiff.getSubColumns().size() != 0) + + if (!columnDiff.getSubColumns().isEmpty()) return columnDiff; else return null; diff --git a/src/org/apache/cassandra/service/ReadRepairManager.java b/src/org/apache/cassandra/service/ReadRepairManager.java index 7299581c41..0b62ebf77c 100644 --- a/src/org/apache/cassandra/service/ReadRepairManager.java +++ b/src/org/apache/cassandra/service/ReadRepairManager.java @@ -106,18 +106,12 @@ class ReadRepairManager } /* - * This is the fn that should be used to scheule a read repair - * specify a endpoint on whcih the read repair should happen and the row mutaion - * message that has the repaired row. + * Schedules a read repair. + * @param target endpoint on whcih the read repair should happen + * @param rowMutationMessage the row mutation message that has the repaired row. */ public void schedule(EndPoint target, RowMutationMessage rowMutationMessage) { - /* - Message message = new Message(StorageService.getLocalStorageEndPoint(), - StorageService.mutationStage_, - StorageService.readRepairVerbHandler_, new Object[] - { rowMutationMessage }); - */ try { Message message = RowMutationMessage.makeRowMutationMessage(rowMutationMessage, StorageService.readRepairVerbHandler_); @@ -126,7 +120,7 @@ class ReadRepairManager } catch ( IOException ex ) { - logger_.info(LogUtil.throwableToString(ex)); + logger_.error(LogUtil.throwableToString(ex)); } } } diff --git a/src/org/apache/cassandra/service/ReadResponseResolver.java b/src/org/apache/cassandra/service/ReadResponseResolver.java index 0981c894c3..ea899c1d78 100644 --- a/src/org/apache/cassandra/service/ReadResponseResolver.java +++ b/src/org/apache/cassandra/service/ReadResponseResolver.java @@ -126,35 +126,26 @@ public class ReadResponseResolver implements IResponseResolver { retRow.repair(rowList.get(i)); } + // At this point we have the return row . // Now we need to calculate the differnce // so that we can schedule read repairs - for (int i = 0 ; i < rowList.size(); i++) { - // calculate the difference , since retRow is the resolved - // row it can be used as the super set , remember no deletes - // will happen with diff its only for additions so far - // TODO : handle deletes + // since retRow is the resolved row it can be used as the super set Row diffRow = rowList.get(i).diff(retRow); if(diffRow == null) // no repair needs to happen continue; // create the row mutation message based on the diff and schedule a read repair RowMutation rowMutation = new RowMutation(table, key); - Map columnFamilies = diffRow.getColumnFamilyMap(); - Set cfNames = columnFamilies.keySet(); - - for ( String cfName : cfNames ) + for (ColumnFamily cf : diffRow.getColumnFamilies()) { - ColumnFamily cf = columnFamilies.get(cfName); rowMutation.add(cf); } RowMutationMessage rowMutationMessage = new RowMutationMessage(rowMutation); - // schedule the read repair ReadRepairManager.instance().schedule(endPoints.get(i),rowMutationMessage); } - logger_.info("resolve: " + (System.currentTimeMillis() - startTime) - + " ms."); + logger_.info("resolve: " + (System.currentTimeMillis() - startTime) + " ms."); return retRow; } diff --git a/src/org/apache/cassandra/service/StorageProxy.java b/src/org/apache/cassandra/service/StorageProxy.java index d146cbc6bf..f496246247 100644 --- a/src/org/apache/cassandra/service/StorageProxy.java +++ b/src/org/apache/cassandra/service/StorageProxy.java @@ -415,7 +415,7 @@ public class StorageProxy private static Row strongRead(ReadCommand command) throws IOException, TimeoutException { // TODO: throw a thrift exception if we do not have N nodes - + assert !command.isDigestQuery(); ReadCommand readMessageDigestOnly = command.copy(); readMessageDigestOnly.setDigestQuery(true); @@ -464,7 +464,6 @@ public class StorageProxy QuorumResponseHandler quorumResponseHandlerRepair = new QuorumResponseHandler( DatabaseDescriptor.getReplicationFactor(), readResponseResolverRepair); - command.setDigestQuery(false); logger_.info("DigestMismatchException: " + command.key); Message messageRepair = command.makeReadMessage(); MessagingService.getMessagingInstance().sendRR(messageRepair, endPoints,