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

use resolve (moved to CF from CFS) rather than duplicating code in repair to do the 

same thing.  this automatically gets us most of the way to handling tombstones in
read repair.  the rest is minor tweaks to CF.diff and SC.diff.  added tests.
patch by jbellis; reviewed by Jun Rao for #87.

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@768109 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-04-24 00:46:32 +00:00
parent 0133ff232f
commit 2ec1b06e5f
6 changed files with 125 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
src/org/apache/cassandra/db/ColumnFamily.java
View File

@ -311,33 +311,24 @@ public final class ColumnFamily
return markedForDeleteAt > Long.MIN_VALUE;
}
/*
* This function will repair a list of columns
* If there are any columns in the external list which are not present
* internally then they are added ( this might have to change depending on
* how we implement delete repairs).
* Also if there are any columns in teh internal and not in the external
* they are kept intact.
* Else the one with the greatest timestamp is considered latest.
*/
void repair(ColumnFamily columnFamily)
{
for (IColumn column : columnFamily.getAllColumns()) {
addColumn(column);
}
}
/*
* This function will calculate the differnce between 2 column families
* the external input is considered the superset of internal
* so there are no deletes in the diff.
*/
ColumnFamily diff(ColumnFamily columnFamily)
ColumnFamily diff(ColumnFamily cfComposite)
{
ColumnFamily cfDiff = new ColumnFamily(columnFamily.name(), columnFamily.type_);
Map<String, IColumn> columns = columnFamily.getColumns();
Set<String> cNames = columns.keySet();
ColumnFamily cfDiff = new ColumnFamily(cfComposite.name(), cfComposite.type_);
if (cfComposite.getMarkedForDeleteAt() > getMarkedForDeleteAt())
{
cfDiff.delete(cfComposite.getLocalDeletionTime(), cfComposite.getMarkedForDeleteAt());
}
// (don't need to worry about cfNew containing IColumns that are shadowed by
// the delete tombstone, since cfNew was generated by CF.resolve, which
// takes care of those for us.)
Map<String, IColumn> columns = cfComposite.getColumns();
Set<String> cNames = columns.keySet();
for ( String cName : cNames )
{
IColumn columnInternal = columns_.get(cName);
@ -355,7 +346,8 @@ public final class ColumnFamily
}
}
}
if(cfDiff.getColumns().size() != 0)
if (!cfDiff.getColumns().isEmpty() || cfDiff.isMarkedForDelete())
return cfDiff;
else
return null;
@ -437,6 +429,28 @@ public final class ColumnFamily
return type_;
}
/** merge all columnFamilies into a single instance, with only the newest versions of columns preserved. */
static ColumnFamily resolve(List<ColumnFamily> columnFamilies)
{
int size = columnFamilies.size();
if (size == 0)
return null;
// start from nothing so that we don't include potential deleted columns from the first instance
ColumnFamily cf0 = columnFamilies.get(0);
ColumnFamily cf = cf0.cloneMeShallow();
// merge
for (ColumnFamily cf2 : columnFamilies)
{
assert cf.name().equals(cf2.name());
cf.addColumns(cf2);
cf.delete(Math.max(cf.getLocalDeletionTime(), cf2.getLocalDeletionTime()),
Math.max(cf.getMarkedForDeleteAt(), cf2.getMarkedForDeleteAt()));
}
return cf;
}
public static class ColumnFamilySerializer implements ICompactSerializer2<ColumnFamily>
{
/*

26
src/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -580,39 +580,17 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
columnFamilies.add(columnFamily);
}
}
/** merge all columnFamilies into a single instance, with only the newest versions of columns preserved. */
static ColumnFamily resolve(List<ColumnFamily> columnFamilies)
{
int size = columnFamilies.size();
if (size == 0)
return null;
// start from nothing so that we don't include potential deleted columns from the first instance
ColumnFamily cf0 = columnFamilies.get(0);
ColumnFamily cf = cf0.cloneMeShallow();
// merge
for (ColumnFamily cf2 : columnFamilies)
{
assert cf.name().equals(cf2.name());
cf.addColumns(cf2);
cf.delete(Math.max(cf.getLocalDeletionTime(), cf2.getLocalDeletionTime()),
Math.max(cf.getMarkedForDeleteAt(), cf2.getMarkedForDeleteAt()));
}
return cf;
}
/** like resolve, but leaves the resolved CF as the only item in the list */
private static void merge(List<ColumnFamily> columnFamilies)
{
ColumnFamily cf = resolve(columnFamilies);
ColumnFamily cf = ColumnFamily.resolve(columnFamilies);
columnFamilies.clear();
columnFamilies.add(cf);
}
private static ColumnFamily resolveAndRemoveDeleted(List<ColumnFamily> columnFamilies) {
ColumnFamily cf = resolve(columnFamilies);
ColumnFamily cf = ColumnFamily.resolve(columnFamilies);
return removeDeleted(cf);
}

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

@ -26,6 +26,7 @@ import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
import java.util.Set;
import java.util.Arrays;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
@ -110,22 +111,21 @@ public class Row
* what that means is that if there are any differences between the 2 rows then
* this fn will make the current row take the latest changes .
*/
public void repair(Row row)
public void repair(Row rowOther)
{
Map<String, ColumnFamily> columnFamilies = row.getColumnFamilyMap();
Set<String> cfNames = columnFamilies.keySet();
for (String cfName : cfNames)
for (ColumnFamily cfOld : rowOther.getColumnFamilies())
{
ColumnFamily cf = columnFamilies_.get(cfName);
ColumnFamily cf = columnFamilies_.get(cfOld.name());
if (cf == null)
{
cf = new ColumnFamily(cfName, cf.type());
columnFamilies_.put(cfName, cf);
addColumnFamily(cfOld);
}
else
{
columnFamilies_.remove(cf.name());
addColumnFamily(ColumnFamily.resolve(Arrays.asList(cfOld, cf)));
}
cf.repair(columnFamilies.get(cfName));
}
}
/*
@ -136,19 +136,18 @@ public class Row
* 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 rowNew)
public Row diff(Row rowComposite)
{
Row rowDiff = new Row(key_);
for (ColumnFamily cfNew : rowNew.getColumnFamilies())
for (ColumnFamily cfComposite : rowComposite.getColumnFamilies())
{
ColumnFamily cf = columnFamilies_.get(cfNew.name());
ColumnFamily cfDiff = null;
ColumnFamily cf = columnFamilies_.get(cfComposite.name());
if (cf == null)
rowDiff.addColumnFamily(cfNew);
rowDiff.addColumnFamily(cfComposite);
else
{
cfDiff = cf.diff(cfNew);
ColumnFamily cfDiff = cf.diff(cfComposite);
if (cfDiff != null)
rowDiff.addColumnFamily(cfDiff);
}

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

@ -230,11 +230,18 @@ public final class SuperColumn implements IColumn, Serializable
return columns_.size();
}
public IColumn diff(IColumn column)
public IColumn diff(IColumn columnNew)
{
IColumn columnDiff = new SuperColumn(column.name());
IColumn columnDiff = new SuperColumn(columnNew.name());
if (columnNew.getMarkedForDeleteAt() > getMarkedForDeleteAt())
{
((SuperColumn)columnDiff).markForDeleteAt(columnNew.getLocalDeletionTime(), columnNew.getMarkedForDeleteAt());
}
for (IColumn subColumn : column.getSubColumns())
// (don't need to worry about columnNew containing subColumns that are shadowed by
// the delete tombstone, since columnNew was generated by CF.resolve, which
// takes care of those for us.)
for (IColumn subColumn : columnNew.getSubColumns())
{
IColumn columnInternal = columns_.get(subColumn.name());
if(columnInternal == null )
@ -251,7 +258,7 @@ public final class SuperColumn implements IColumn, Serializable
}
}
if (!columnDiff.getSubColumns().isEmpty())
if (!columnDiff.getSubColumns().isEmpty() || columnNew.isMarkedForDelete())
return columnDiff;
else
return null;

2
test/org/apache/cassandra/db/ColumnFamilyStoreTest.java
View File

@ -257,7 +257,7 @@ public class ColumnFamilyStoreTest extends ServerTest
List<ColumnFamily> families = store.getColumnFamilies("key1", "Super1", new IdentityFilter());
assert families.get(0).getAllColumns().first().getMarkedForDeleteAt() == 1; // delete marker, just added
assert !families.get(1).getAllColumns().first().isMarkedForDelete(); // flushed old version
ColumnFamily resolved = ColumnFamilyStore.resolve(families);
ColumnFamily resolved = ColumnFamily.resolve(families);
assert resolved.getAllColumns().first().getMarkedForDeleteAt() == 1;
Collection<IColumn> subColumns = resolved.getAllColumns().first().getSubColumns();
assert subColumns.size() == 1;

61
test/org/apache/cassandra/db/RowTest.java Normal file
View File

@ -0,0 +1,61 @@
package org.apache.cassandra.db;
import java.util.Arrays;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
public class RowTest
{
@Test
public void testDiffColumnFamily()
{
ColumnFamily cf1 = new ColumnFamily("Standard1", "Standard");
cf1.addColumn("one", "onev".getBytes(), 0);
ColumnFamily cf2 = new ColumnFamily("Standard1", "Standard");
cf2.delete(0, 0);
ColumnFamily cfDiff = cf1.diff(cf2);
assertEquals(cfDiff.getColumns().size(), 0);
assertEquals(cfDiff.getMarkedForDeleteAt(), 0);
}
@Test
public void testDiffSuperColumn()
{
SuperColumn sc1 = new SuperColumn("one");
sc1.addColumn("subcolumn", new Column("subcolumn", "A".getBytes(), 0));
SuperColumn sc2 = new SuperColumn("one");
sc2.markForDeleteAt(0, 0);
SuperColumn scDiff = (SuperColumn)sc1.diff(sc2);
assertEquals(scDiff.getSubColumns().size(), 0);
assertEquals(scDiff.getMarkedForDeleteAt(), 0);
}
@Test
public void testRepair()
{
Row row1 = new Row();
ColumnFamily cf1 = new ColumnFamily("Standard1", "Standard");
cf1.addColumn("one", "A".getBytes(), 0);
row1.addColumnFamily(cf1);
Row row2 = new Row();
ColumnFamily cf2 = new ColumnFamily("Standard1", "Standard");
cf2.addColumn("one", "B".getBytes(), 1);
cf2.addColumn("two", "C".getBytes(), 1);
ColumnFamily cf3 = new ColumnFamily("Standard2", "Standard");
cf3.addColumn("three", "D".getBytes(), 1);
row2.addColumnFamily(cf2);
row2.addColumnFamily(cf3);
row1.repair(row2);
cf1 = row1.getColumnFamily("Standard1");
assert Arrays.equals(cf1.getColumn("one").value(), "B".getBytes());
assert Arrays.equals(cf2.getColumn("two").value(), "C".getBytes());
assert row1.getColumnFamily("Standard2") != null;
}
}