natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

clean up Read Repair code to use non-deprecated APIs. 

patch by jbellis; reviewed by Jun Rao for #87.

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@768108 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-04-24 00:46:19 +00:00
parent 3c6fcf8d65
commit 0133ff232f
5 changed files with 23 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
src/org/apache/cassandra/db/Row.java
View File

@ -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<String, ColumnFamily> columnFamilies = row.getColumnFamilyMap();
Set<String> 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()

8
src/org/apache/cassandra/db/SuperColumn.java
View File

@ -232,10 +232,9 @@ public final class SuperColumn implements IColumn, Serializable
public IColumn diff(IColumn column)
{
IColumn columnDiff = new SuperColumn(column.name());
Collection<IColumn> 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;

14
src/org/apache/cassandra/service/ReadRepairManager.java
View File

@ -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));
}
}
}

17
src/org/apache/cassandra/service/ReadResponseResolver.java
View File

@ -126,35 +126,26 @@ public class ReadResponseResolver implements IResponseResolver<Row>
{
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<String, ColumnFamily> columnFamilies = diffRow.getColumnFamilyMap();
Set<String> 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;
}

3
src/org/apache/cassandra/service/StorageProxy.java
View File

@ -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<Row> quorumResponseHandlerRepair = new QuorumResponseHandler<Row>(
DatabaseDescriptor.getReplicationFactor(),
readResponseResolverRepair);
command.setDigestQuery(false);
logger_.info("DigestMismatchException: " + command.key);
Message messageRepair = command.makeReadMessage();
MessagingService.getMessagingInstance().sendRR(messageRepair, endPoints,