From dd4f4be2b426361b1c50b784f8e3c6557e765497 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Fri, 10 Jul 2009 17:49:18 +0000 Subject: [PATCH] add IdentityQueryFilter and finish removing IFilter. patch by jbellis; reviewed by Jun Rao for CASSANDRA-287 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@793055 13f79535-47bb-0310-9956-ffa450edef68 --- .../cassandra/db/AbstractColumnFactory.java | 7 +- src/java/org/apache/cassandra/db/Column.java | 77 +-------- .../org/apache/cassandra/db/ColumnFamily.java | 83 +--------- .../cassandra/db/ColumnFamilyStore.java | 155 ++---------------- .../cassandra/db/ColumnReadCommand.java | 11 +- .../CompactSerializerInvocationHandler.java | 7 +- .../cassandra/db/HintedHandOffManager.java | 3 +- .../cassandra/db/ICompactSerializer2.java | 65 -------- src/java/org/apache/cassandra/db/IFilter.java | 33 ---- .../apache/cassandra/db/IdentityFilter.java | 44 ----- .../org/apache/cassandra/db/Memtable.java | 42 ----- .../org/apache/cassandra/db/ReadCommand.java | 13 +- .../cassandra/db/SliceFromReadCommand.java | 2 +- .../org/apache/cassandra/db/SuperColumn.java | 114 +------------ .../org/apache/cassandra/db/SystemTable.java | 6 +- src/java/org/apache/cassandra/db/Table.java | 72 +++----- .../db/filter/IdentityQueryFilter.java | 20 +++ .../org/apache/cassandra/io/SequenceFile.java | 70 -------- .../cassandra/db/ColumnFamilyStoreTest.java | 3 +- .../cassandra/db/RemoveColumnFamilyTest.java | 3 +- .../db/RemoveColumnFamilyWithFlush1Test.java | 3 +- .../db/RemoveColumnFamilyWithFlush2Test.java | 3 +- .../apache/cassandra/db/RemoveColumnTest.java | 5 +- .../cassandra/db/RemoveSubColumnTest.java | 3 +- .../cassandra/db/RemoveSuperColumnTest.java | 23 +-- .../org/apache/cassandra/db/TableTest.java | 8 +- 26 files changed, 125 insertions(+), 750 deletions(-) delete mode 100644 src/java/org/apache/cassandra/db/ICompactSerializer2.java delete mode 100644 src/java/org/apache/cassandra/db/IFilter.java delete mode 100644 src/java/org/apache/cassandra/db/IdentityFilter.java create mode 100644 src/java/org/apache/cassandra/db/filter/IdentityQueryFilter.java diff --git a/src/java/org/apache/cassandra/db/AbstractColumnFactory.java b/src/java/org/apache/cassandra/db/AbstractColumnFactory.java index 4048f9933c..f50afbb8de 100644 --- a/src/java/org/apache/cassandra/db/AbstractColumnFactory.java +++ b/src/java/org/apache/cassandra/db/AbstractColumnFactory.java @@ -22,6 +22,7 @@ import java.util.HashMap; import java.util.Map; import java.util.StringTokenizer; +import org.apache.cassandra.io.ICompactSerializer; /** @@ -51,7 +52,7 @@ abstract class AbstractColumnFactory public abstract IColumn createColumn(String name, byte[] value); public abstract IColumn createColumn(String name, byte[] value, long timestamp); public abstract IColumn createColumn(String name, byte[] value, long timestamp, boolean deleted); - public abstract ICompactSerializer2 createColumnSerializer(); + public abstract ICompactSerializer createColumnSerializer(); } class ColumnFactory extends AbstractColumnFactory @@ -75,7 +76,7 @@ class ColumnFactory extends AbstractColumnFactory return new Column(name, value, timestamp, deleted); } - public ICompactSerializer2 createColumnSerializer() + public ICompactSerializer createColumnSerializer() { return Column.serializer(); } @@ -130,7 +131,7 @@ class SuperColumnFactory extends AbstractColumnFactory return superColumn; } - public ICompactSerializer2 createColumnSerializer() + public ICompactSerializer createColumnSerializer() { return SuperColumn.serializer(); } diff --git a/src/java/org/apache/cassandra/db/Column.java b/src/java/org/apache/cassandra/db/Column.java index e6a109f6af..3e1cede7cb 100644 --- a/src/java/org/apache/cassandra/db/Column.java +++ b/src/java/org/apache/cassandra/db/Column.java @@ -22,12 +22,12 @@ import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.IOException; import java.util.Collection; -import java.util.Map; import java.nio.ByteBuffer; import org.apache.commons.lang.ArrayUtils; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.io.ICompactSerializer; /** @@ -212,7 +212,7 @@ public final class Column implements IColumn } } -class ColumnSerializer implements ICompactSerializer2 +class ColumnSerializer implements ICompactSerializer { public void serialize(IColumn column, DataOutputStream dos) throws IOException { @@ -223,81 +223,14 @@ class ColumnSerializer implements ICompactSerializer2 dos.write(column.value()); } - private IColumn defreeze(DataInputStream dis, String name) throws IOException + public IColumn deserialize(DataInputStream dis) throws IOException { - IColumn column = null; + String name = dis.readUTF(); boolean delete = dis.readBoolean(); long ts = dis.readLong(); int size = dis.readInt(); byte[] value = new byte[size]; dis.readFully(value); - column = new Column(name, value, ts, delete); - return column; - } - - public IColumn deserialize(DataInputStream dis) throws IOException - { - String name = dis.readUTF(); - return defreeze(dis, name); - } - - /** - * Here we need to get the column and apply the filter. - */ - public IColumn deserialize(DataInputStream dis, IFilter filter) throws IOException - { - assert dis.available() > 0; - - String name = dis.readUTF(); - IColumn column = new Column(name); - column = filter.filter(column, dis); - if ( column != null ) - { - column = defreeze(dis, name); - } - else - { - /* Skip a boolean and the timestamp */ - dis.skip(DBConstants.boolSize_ + DBConstants.tsSize_); - int size = dis.readInt(); - dis.skip(size); - } - return column; - } - - /** - * We know the name of the column here so just return it. - * Filter is pretty much useless in this call and is ignored. - */ - public IColumn deserialize(DataInputStream dis, String columnName, IFilter filter) throws IOException - { - assert dis.available() > 0; - IColumn column = null; - String name = dis.readUTF(); - if ( name.equals(columnName) ) - { - column = defreeze(dis, name); - } - else - { - /* Skip a boolean and the timestamp */ - dis.skip(DBConstants.boolSize_ + DBConstants.tsSize_); - int size = dis.readInt(); - dis.skip(size); - } - return column; - } - - public void skip(DataInputStream dis) throws IOException - { - /* read the column name */ - dis.readUTF(); - /* boolean indicating if the column is deleted */ - dis.readBoolean(); - /* timestamp associated with the column */ - dis.readLong(); - /* size of the column */ - int size = dis.readInt(); - dis.skip(size); + return new Column(name, value, ts, delete); } } diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java index 0554f4d680..d7661725ff 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamily.java +++ b/src/java/org/apache/cassandra/db/ColumnFamily.java @@ -37,6 +37,7 @@ import org.apache.log4j.Logger; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.io.ICompactSerializer; /** * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com ) @@ -44,7 +45,7 @@ import org.apache.cassandra.utils.FBUtilities; public final class ColumnFamily { /* The column serializer for this Column Family. Create based on config. */ - private static ICompactSerializer2 serializer_; + private static ICompactSerializer serializer_; public static final short utfPrefix_ = 2; private static Logger logger_ = Logger.getLogger( ColumnFamily.class ); @@ -64,7 +65,7 @@ public final class ColumnFamily indexTypes_.put("Time", "Time"); } - public static ICompactSerializer2 serializer() + public static ICompactSerializer serializer() { return serializer_; } @@ -73,9 +74,9 @@ public final class ColumnFamily * This method returns the serializer whose methods are * preprocessed by a dynamic proxy. */ - public static ICompactSerializer2 serializerWithIndexes() + public static ICompactSerializer serializerWithIndexes() { - return (ICompactSerializer2)Proxy.newProxyInstance( ColumnFamily.class.getClassLoader(), new Class[]{ICompactSerializer2.class}, new CompactSerializerInvocationHandler(serializer_) ); + return (ICompactSerializer)Proxy.newProxyInstance( ColumnFamily.class.getClassLoader(), new Class[]{ICompactSerializer.class}, new CompactSerializerInvocationHandler(serializer_) ); } public static String getColumnType(String key) @@ -112,7 +113,7 @@ public final class ColumnFamily private String name_; - private transient ICompactSerializer2 columnSerializer_; + private transient ICompactSerializer columnSerializer_; private long markedForDeleteAt = Long.MIN_VALUE; private int localDeletionTime = Integer.MIN_VALUE; private AtomicInteger size_ = new AtomicInteger(0); @@ -165,7 +166,7 @@ public final class ColumnFamily } } - public ICompactSerializer2 getColumnSerializer() + public ICompactSerializer getColumnSerializer() { return columnSerializer_; } @@ -449,7 +450,7 @@ public final class ColumnFamily return cf; } - public static class ColumnFamilySerializer implements ICompactSerializer2 + public static class ColumnFamilySerializer implements ICompactSerializer { /* * We are going to create indexes, and write out that information as well. The format @@ -492,22 +493,12 @@ public final class ColumnFamily } } - /* - * Use this method to create a bare bones Column Family. This column family - * does not have any of the Column information. - */ - private ColumnFamily defreezeColumnFamily(DataInputStream dis) throws IOException + public ColumnFamily deserialize(DataInputStream dis) throws IOException { ColumnFamily cf = new ColumnFamily(dis.readUTF(), dis.readUTF(), ColumnComparatorFactory.ComparatorType.values()[dis.readInt()]); cf.delete(dis.readInt(), dis.readLong()); - return cf; - } - - public ColumnFamily deserialize(DataInputStream dis) throws IOException - { - ColumnFamily cf = defreezeColumnFamily(dis); int size = dis.readInt(); IColumn column; for (int i = 0; i < size; ++i) @@ -517,62 +508,6 @@ public final class ColumnFamily } return cf; } - - /* - * This version of deserialize is used when we need a specific set if columns for - * a column family specified in the name cfName parameter. - */ - public ColumnFamily deserialize(DataInputStream dis, IFilter filter) throws IOException - { - ColumnFamily cf = defreezeColumnFamily(dis); - int size = dis.readInt(); - IColumn column = null; - for ( int i = 0; i < size; ++i ) - { - column = cf.getColumnSerializer().deserialize(dis, filter); - if(column != null) - { - cf.addColumn(column); - } - } - return cf; - } - - /* - * Deserialize a particular column or super column or the entire columnfamily given a : seprated name - * name could be of the form cf:superColumn:column or cf:column or cf - */ - public ColumnFamily deserialize(DataInputStream dis, String name, IFilter filter) throws IOException - { - String[] names = RowMutation.getColumnAndColumnFamily(name); - String columnName = ""; - if ( names.length == 1 ) - return deserialize(dis, filter); - if( names.length == 2 ) - columnName = names[1]; - if( names.length == 3 ) - columnName = names[1]+ ":" + names[2]; - - ColumnFamily cf = defreezeColumnFamily(dis); - /* read the number of columns */ - int size = dis.readInt(); - for ( int i = 0; i < size; ++i ) - { - IColumn column = cf.getColumnSerializer().deserialize(dis, columnName, filter); - if ( column != null ) - { - cf.addColumn(column); - break; - } - } - return cf; - } - - public void skip(DataInputStream dis) throws IOException - { - throw new UnsupportedOperationException("This operation is not yet supported."); - } } - } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 89e1d2740b..b6f1df6ca9 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -508,116 +508,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean writeStats_.add(System.currentTimeMillis() - start); } - public ColumnFamily getColumnFamily(String key, String columnFamilyColumn, IFilter filter) throws IOException - { - long start = System.currentTimeMillis(); - List columnFamilies = getColumnFamilies(key, columnFamilyColumn, filter); - ColumnFamily cf = resolveAndRemoveDeleted(columnFamilies); - readStats_.add(System.currentTimeMillis() - start); - return cf; - } - - public ColumnFamily getColumnFamily(String key, String columnFamilyColumn, IFilter filter, int gcBefore) throws IOException - { - long start = System.currentTimeMillis(); - List columnFamilies = getColumnFamilies(key, columnFamilyColumn, filter); - ColumnFamily cf = removeDeleted(ColumnFamily.resolve(columnFamilies), gcBefore); - readStats_.add(System.currentTimeMillis() - start); - return cf; - } - - /** - * Get the column family in the most efficient order. - * 1. Memtable - * 2. Sorted list of files - */ - List getColumnFamilies(String key, String columnFamilyColumn, IFilter filter) throws IOException - { - List columnFamilies = new ArrayList(); - /* Get the ColumnFamily from Memtable */ - getColumnFamilyFromCurrentMemtable(key, columnFamilyColumn, filter, columnFamilies); - /* Check if MemtableManager has any historical information */ - getUnflushedColumnFamily(key, columnFamily_, columnFamilyColumn, filter, columnFamilies); - long start = System.currentTimeMillis(); - getColumnFamilyFromDisk(key, columnFamilyColumn, columnFamilies, filter); - diskReadStats_.add(System.currentTimeMillis() - start); - - return columnFamilies; - } - - /** - * Fetch from disk files and go in sorted order to be efficient - * This function exits as soon as the required data is found. - * - * @param key - * @param cf - * @param columnFamilies - * @param filter - * @throws IOException - */ - private void getColumnFamilyFromDisk(String key, String cf, List columnFamilies, IFilter filter) throws IOException - { - sstableLock_.readLock().lock(); - try - { - for (SSTableReader sstable : ssTables_.values()) - { - ColumnFamily columnFamily = null; - try - { - columnFamily = fetchColumnFamily(key, cf, filter, sstable); - } - catch (IOException e) - { - // annotate exception w/ more information about context - throw new IOException("Error fetching " + key + ":" + cf + " from " + sstable, e); - } - if (columnFamily != null) - { - columnFamilies.add(columnFamily); - } - } - } - finally - { - sstableLock_.readLock().unlock(); - } - } - - private ColumnFamily fetchColumnFamily(String key, String cf, IFilter filter, SSTableReader ssTable) throws IOException - { - DataInputBuffer bufIn; - bufIn = filter.next(key, cf, ssTable); - if (bufIn.getLength() == 0) - { - return null; - } - ColumnFamily columnFamily = ColumnFamily.serializer().deserialize(bufIn, cf, filter); - if (columnFamily == null) - { - return null; - } - return columnFamily; - } - - private void getColumnFamilyFromCurrentMemtable(String key, String cf, IFilter filter, List columnFamilies) - { - ColumnFamily columnFamily; - memtableLock_.readLock().lock(); - try - { - columnFamily = memtable_.getLocalCopy(key, cf, filter); - } - finally - { - memtableLock_.readLock().unlock(); - } - if (columnFamily != null) - { - columnFamilies.add(columnFamily); - } - } - /** * like resolve, but leaves the resolved CF as the only item in the list */ @@ -1404,26 +1294,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean return memtables; } - /* - * Retrieve column family from the list of Memtables that have been - * submitted for flush but have not yet been flushed. - * It also filters out unneccesary columns based on the passed in filter. - */ - void getUnflushedColumnFamily(String key, String cfName, String cf, IFilter filter, List columnFamilies) - { - List memtables = getUnflushedMemtables(cfName); - Collections.sort(memtables); - int size = memtables.size(); - for ( int i = size - 1; i >= 0; --i ) - { - ColumnFamily columnFamily = memtables.get(i).getLocalCopy(key, cf, filter); - if ( columnFamily != null ) - { - columnFamilies.add(columnFamily); - } - } - } - /* Submit memtables to be flushed to disk */ public static void submitFlush(final Memtable memtable, final CommitLog.CommitLogContext cLogCtx) { @@ -1554,11 +1424,11 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean } /** - * get a list of columns starting from a given column, in a specified order - * only the latest version of a column is returned + * get a list of columns starting from a given column, in a specified order. + * only the latest version of a column is returned. + * @return null if there is no data and no tombstones; otherwise a ColumnFamily */ - public ColumnFamily getColumnFamily(QueryFilter filter) - throws IOException + public ColumnFamily getColumnFamily(QueryFilter filter) throws IOException { String[] values = RowMutation.getColumnAndColumnFamily(filter.columnFamilyColumn); @@ -1567,9 +1437,12 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean { QueryFilter nameFilter = new NamesQueryFilter(filter.key, values[0], values[1]); ColumnFamily cf = getColumnFamily(nameFilter); - for (IColumn column : cf.getAllColumns()) + if (cf != null) { - filter.filterSuperColumn((SuperColumn) column); + for (IColumn column : cf.getAllColumns()) + { + filter.filterSuperColumn((SuperColumn) column); + } } return removeDeleted(cf); } @@ -1609,21 +1482,17 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean for (SSTableReader sstable : sstables) { iter = filter.getSSTableColumnIterator(sstable); - if (iter.hasNext()) + if (iter.hasNext()) // initializes iter.CF { returnCF.delete(iter.getColumnFamily()); - iterators.add(iter); - } - else - { - iter.close(); } + iterators.add(iter); } Comparator comparator = filter.getColumnComparator(); Iterator collated = IteratorUtils.collatedIterator(comparator, iterators); if (!collated.hasNext()) - return ColumnFamily.create(table_, columnFamily_); + return null; filter.collectColumns(returnCF, collated); diff --git a/src/java/org/apache/cassandra/db/ColumnReadCommand.java b/src/java/org/apache/cassandra/db/ColumnReadCommand.java index ff14c113ef..7353552955 100644 --- a/src/java/org/apache/cassandra/db/ColumnReadCommand.java +++ b/src/java/org/apache/cassandra/db/ColumnReadCommand.java @@ -21,6 +21,8 @@ package org.apache.cassandra.db; import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.IOException; +import java.util.TreeSet; +import java.util.Arrays; public class ColumnReadCommand extends ReadCommand { @@ -49,7 +51,14 @@ public class ColumnReadCommand extends ReadCommand @Override public Row getRow(Table table) throws IOException { - return table.getRow(key, columnFamilyColumn); + String[] values = RowMutation.getColumnAndColumnFamily(columnFamilyColumn); + assert values.length > 1 && values.length <= 3; + if (values.length == 2) + { + return table.getRow(key, values[0], new TreeSet(Arrays.asList(values[1]))); + } + assert values.length == 3 : columnFamilyColumn; + return table.getRow(key, values[0] + ":" + values[1], new TreeSet(Arrays.asList(values[2]))); } @Override diff --git a/src/java/org/apache/cassandra/db/CompactSerializerInvocationHandler.java b/src/java/org/apache/cassandra/db/CompactSerializerInvocationHandler.java index b66772e935..845a3f395b 100644 --- a/src/java/org/apache/cassandra/db/CompactSerializerInvocationHandler.java +++ b/src/java/org/apache/cassandra/db/CompactSerializerInvocationHandler.java @@ -22,19 +22,20 @@ import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import org.apache.cassandra.io.DataOutputBuffer; +import org.apache.cassandra.io.ICompactSerializer; /* * This is the abstraction that pre-processes calls to implementations - * of the ICompactSerializer2 serialize() via dynamic proxies. + * of the ICompactSerializer serialize() via dynamic proxies. * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com ) */ public class CompactSerializerInvocationHandler implements InvocationHandler { - private ICompactSerializer2 serializer_; + private ICompactSerializer serializer_; - public CompactSerializerInvocationHandler(ICompactSerializer2 serializer) + public CompactSerializerInvocationHandler(ICompactSerializer serializer) { serializer_ = serializer; } diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index fb64878da0..39406e5ee3 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -37,6 +37,7 @@ import org.apache.cassandra.net.EndPoint; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.*; +import org.apache.cassandra.db.filter.IdentityQueryFilter; /** @@ -159,7 +160,7 @@ public class HintedHandOffManager // 7. I guess we are done for (String tableName : DatabaseDescriptor.getTables()) { - ColumnFamily hintColumnFamily = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(tableName, HINTS_CF, new IdentityFilter()), Integer.MAX_VALUE); + ColumnFamily hintColumnFamily = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(new IdentityQueryFilter(tableName, HINTS_CF)), Integer.MAX_VALUE); if (hintColumnFamily == null) { continue; diff --git a/src/java/org/apache/cassandra/db/ICompactSerializer2.java b/src/java/org/apache/cassandra/db/ICompactSerializer2.java deleted file mode 100644 index de57a50cf3..0000000000 --- a/src/java/org/apache/cassandra/db/ICompactSerializer2.java +++ /dev/null @@ -1,65 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.cassandra.db; - -import java.io.DataInputStream; -import java.io.IOException; -import java.util.List; - -import org.apache.cassandra.io.ICompactSerializer; -import org.apache.cassandra.utils.BloomFilter; - - -/** - * This interface is an extension of the ICompactSerializer which allows for partial deserialization - * of a type. - * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com ) - */ - -public interface ICompactSerializer2 extends ICompactSerializer -{ - /** - * Returns an instance of an IColumn which contains only the - * columns that are required. This is specified in the columnNames - * argument. - * - * @param dis DataInput from which we need to deserialize. - * @throws IOException - * @return type which contains the specified items. - */ - public T deserialize(DataInputStream dis, IFilter filter) throws IOException; - - /** - * This method is used to deserialize just the specified field from - * the serialized stream. - * - * @param dis DataInput from which we need to deserialize. - * @param name name of the desired field. - * @throws IOException - * @return the deserialized type. - */ - public T deserialize(DataInputStream dis, String name, IFilter filter) throws IOException; - - /** - * - * @param dis - * @throws IOException - */ - public void skip(DataInputStream dis) throws IOException; -} diff --git a/src/java/org/apache/cassandra/db/IFilter.java b/src/java/org/apache/cassandra/db/IFilter.java deleted file mode 100644 index 87963bab84..0000000000 --- a/src/java/org/apache/cassandra/db/IFilter.java +++ /dev/null @@ -1,33 +0,0 @@ -/* -* Licensed to the Apache Software Foundation (ASF) under one -* or more contributor license agreements. See the NOTICE file -* distributed with this work for additional information -* regarding copyright ownership. The ASF licenses this file -* to you under the Apache License, Version 2.0 (the -* "License"); you may not use this file except in compliance -* with the License. You may obtain a copy of the License at -* -* http://www.apache.org/licenses/LICENSE-2.0 -* -* Unless required by applicable law or agreed to in writing, -* software distributed under the License is distributed on an -* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -* KIND, either express or implied. See the License for the -* specific language governing permissions and limitations -* under the License. -*/ -package org.apache.cassandra.db; - -import java.io.DataInputStream; -import java.io.IOException; - -import org.apache.cassandra.io.DataInputBuffer; -import org.apache.cassandra.io.SSTableReader; - - -public interface IFilter -{ - public ColumnFamily filter(String cfName, ColumnFamily cf); - public IColumn filter(IColumn column, DataInputStream dis) throws IOException; - public DataInputBuffer next(String key, String cf, SSTableReader ssTable) throws IOException; -} diff --git a/src/java/org/apache/cassandra/db/IdentityFilter.java b/src/java/org/apache/cassandra/db/IdentityFilter.java deleted file mode 100644 index deb3e4b9b0..0000000000 --- a/src/java/org/apache/cassandra/db/IdentityFilter.java +++ /dev/null @@ -1,44 +0,0 @@ -/* -* Licensed to the Apache Software Foundation (ASF) under one -* or more contributor license agreements. See the NOTICE file -* distributed with this work for additional information -* regarding copyright ownership. The ASF licenses this file -* to you under the Apache License, Version 2.0 (the -* "License"); you may not use this file except in compliance -* with the License. You may obtain a copy of the License at -* -* http://www.apache.org/licenses/LICENSE-2.0 -* -* Unless required by applicable law or agreed to in writing, -* software distributed under the License is distributed on an -* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -* KIND, either express or implied. See the License for the -* specific language governing permissions and limitations -* under the License. -*/ -package org.apache.cassandra.db; - -import java.io.DataInputStream; -import java.io.IOException; - -import org.apache.cassandra.io.DataInputBuffer; -import org.apache.cassandra.io.SSTableReader; - - -public class IdentityFilter implements IFilter -{ - public ColumnFamily filter(String cfString, ColumnFamily columnFamily) - { - return columnFamily; - } - - public IColumn filter(IColumn column, DataInputStream dis) throws IOException - { - return column; - } - - public DataInputBuffer next(String key, String cf, SSTableReader ssTable) throws IOException - { - return ssTable.next(key, cf); - } -} diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index d632fb7021..37a323727c 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -204,48 +204,6 @@ public class Memtable implements Comparable return builder.toString(); } - ColumnFamily getLocalCopy(String key, String columnFamilyColumn, IFilter filter) - { - String[] values = RowMutation.getColumnAndColumnFamily(columnFamilyColumn); - ColumnFamily columnFamily = null; - if(values.length == 1 ) - { - columnFamily = columnFamilies_.get(key); - } - else - { - ColumnFamily cFamily = columnFamilies_.get(key); - if (cFamily == null) return null; - - if (values.length == 2) { - IColumn column = cFamily.getColumn(values[1]); // super or normal column - if (column != null ) - { - columnFamily = cFamily.cloneMeShallow(); - columnFamily.addColumn(column); - } - } - else - { - assert values.length == 3; - SuperColumn superColumn = (SuperColumn)cFamily.getColumn(values[1]); - if (superColumn != null) - { - IColumn subColumn = superColumn.getSubColumn(values[2]); - if (subColumn != null) - { - columnFamily = cFamily.cloneMeShallow(); - SuperColumn container = superColumn.cloneMeShallow(); - container.addColumn(subColumn); - columnFamily.addColumn(container); - } - } - } - } - /* Filter unnecessary data from the column based on the provided filter */ - return filter.filter(columnFamilyColumn, columnFamily); - } - void flush(CommitLog.CommitLogContext cLogCtx) throws IOException { logger_.info("Flushing " + this); diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 35a6c5a02e..bd2e738146 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -33,13 +33,10 @@ import org.apache.cassandra.service.StorageService; public abstract class ReadCommand { public static final String DO_REPAIR = "READ-REPAIR"; - public static final byte CMD_TYPE_GET_ROW=1; - public static final byte CMD_TYPE_GET_COLUMN=2; - public static final byte CMD_TYPE_GET_SLICE_BY_NAMES=3; - public static final byte CMD_TYPE_GET_COLUMNS_SINCE=4; - public static final byte CMD_TYPE_GET_SLICE=5; - public static final byte CMD_TYPE_GET_SLICE_BY_RANGE = 6; - public static final byte CMD_TYPE_GET_SLICE_FROM=7; + public static final byte CMD_TYPE_GET_COLUMN=1; + public static final byte CMD_TYPE_GET_SLICE_BY_NAMES=2; + public static final byte CMD_TYPE_GET_COLUMNS_SINCE=3; + public static final byte CMD_TYPE_GET_SLICE =4; public static final String EMPTY_CF = ""; @@ -95,7 +92,7 @@ class ReadCommandSerializer implements ICompactSerializer CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_COLUMN, new ColumnReadCommandSerializer()); CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_SLICE_BY_NAMES, new SliceByNamesReadCommandSerializer()); CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_COLUMNS_SINCE, new ColumnsSinceReadCommandSerializer()); - CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_SLICE_FROM, new SliceFromReadCommandSerializer()); + CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_SLICE, new SliceFromReadCommandSerializer()); } diff --git a/src/java/org/apache/cassandra/db/SliceFromReadCommand.java b/src/java/org/apache/cassandra/db/SliceFromReadCommand.java index d38c858496..a42f6be954 100644 --- a/src/java/org/apache/cassandra/db/SliceFromReadCommand.java +++ b/src/java/org/apache/cassandra/db/SliceFromReadCommand.java @@ -31,7 +31,7 @@ public class SliceFromReadCommand extends ReadCommand public SliceFromReadCommand(String table, String key, String columnFamilyColumn, String start, String finish, boolean isAscending, int offset, int count) { - super(table, key, CMD_TYPE_GET_SLICE_FROM); + super(table, key, CMD_TYPE_GET_SLICE); this.columnFamilyColumn = columnFamilyColumn; this.start = start; this.finish = finish; diff --git a/src/java/org/apache/cassandra/db/SuperColumn.java b/src/java/org/apache/cassandra/db/SuperColumn.java index 63c49839b5..0212857543 100644 --- a/src/java/org/apache/cassandra/db/SuperColumn.java +++ b/src/java/org/apache/cassandra/db/SuperColumn.java @@ -24,7 +24,6 @@ import java.io.IOException; import java.io.Serializable; import java.util.Collection; import java.util.Set; -import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.lang.ArrayUtils; @@ -32,6 +31,7 @@ import org.apache.commons.lang.StringUtils; import org.apache.log4j.Logger; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.io.ICompactSerializer; /** * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com ) @@ -319,7 +319,7 @@ public final class SuperColumn implements IColumn, Serializable } } -class SuperColumnSerializer implements ICompactSerializer2 +class SuperColumnSerializer implements ICompactSerializer { public void serialize(IColumn column, DataOutputStream dos) throws IOException { @@ -339,37 +339,11 @@ class SuperColumnSerializer implements ICompactSerializer2 } } - /* - * Use this method to create a bare bones Super Column. This super column - * does not have any of the Column information. - */ - private SuperColumn defreezeSuperColumn(DataInputStream dis) throws IOException + public IColumn deserialize(DataInputStream dis) throws IOException { String name = dis.readUTF(); SuperColumn superColumn = new SuperColumn(name); superColumn.markForDeleteAt(dis.readInt(), dis.readLong()); - return superColumn; - } - - public IColumn deserialize(DataInputStream dis) throws IOException - { - SuperColumn superColumn = defreezeSuperColumn(dis); - fillSuperColumn(superColumn, dis); - return superColumn; - } - - public void skip(DataInputStream dis) throws IOException - { - defreezeSuperColumn(dis); - /* read the number of columns stored */ - dis.readInt(); - /* read the size of all columns to skip */ - int size = dis.readInt(); - dis.skip(size); - } - - private void fillSuperColumn(IColumn superColumn, DataInputStream dis) throws IOException - { assert dis.available() > 0; /* read the number of columns */ @@ -381,87 +355,7 @@ class SuperColumnSerializer implements ICompactSerializer2 IColumn subColumn = Column.serializer().deserialize(dis); superColumn.addColumn(subColumn); } - } - - public IColumn deserialize(DataInputStream dis, IFilter filter) throws IOException - { - assert dis.available() > 0; - - IColumn superColumn = defreezeSuperColumn(dis); - superColumn = filter.filter(superColumn, dis); - if(superColumn != null) - { - fillSuperColumn(superColumn, dis); - return superColumn; - } - else - { - /* read the number of columns stored */ - dis.readInt(); - /* read the size of all columns to skip */ - int size = dis.readInt(); - dis.skip(size); - return null; - } - } - - /* - * Deserialize a particular column since the name is in the form of - * superColumn:column. - */ - public IColumn deserialize(DataInputStream dis, String name, IFilter filter) throws IOException - { - assert dis.available() > 0; - - String[] names = RowMutation.getColumnAndColumnFamily(name); - if ( names.length == 1 ) - { - IColumn superColumn = defreezeSuperColumn(dis); - if(name.equals(superColumn.name())) - { - /* read the number of columns stored */ - int size = dis.readInt(); - /* read the size of all columns */ - dis.readInt(); - IColumn column = null; - for ( int i = 0; i < size; ++i ) - { - column = Column.serializer().deserialize(dis, filter); - if(column != null) - { - superColumn.addColumn(column); - } - } - return superColumn; - } - else - { - /* read the number of columns stored */ - dis.readInt(); - /* read the size of all columns to skip */ - int size = dis.readInt(); - dis.skip(size); - return null; - } - } - - SuperColumn superColumn = defreezeSuperColumn(dis); - int size = dis.readInt(); - /* skip the size of the columns */ - dis.readInt(); - if ( size > 0 ) - { - for ( int i = 0; i < size; ++i ) - { - IColumn subColumn = Column.serializer().deserialize(dis, names[1], filter); - if ( subColumn != null ) - { - superColumn.addColumn(subColumn); - break; - } - } - } - return superColumn; } + } diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java index b12a6a4650..98fbb2681b 100644 --- a/src/java/org/apache/cassandra/db/SystemTable.java +++ b/src/java/org/apache/cassandra/db/SystemTable.java @@ -26,6 +26,8 @@ import org.apache.cassandra.service.StorageService; import org.apache.cassandra.dht.Token; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.utils.BasicUtilities; +import org.apache.cassandra.db.filter.IdentityQueryFilter; +import org.apache.cassandra.db.filter.NamesQueryFilter; /** * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com ) @@ -47,7 +49,7 @@ public class SystemTable IPartitioner p = StorageService.getPartitioner(); Table table = Table.open(Table.SYSTEM_TABLE); /* Retrieve the "LocationInfo" column family */ - ColumnFamily cf = table.getColumnFamilyStore(LOCATION_CF).getColumnFamily(LOCATION_KEY, LOCATION_KEY + ":" + TOKEN, new IdentityFilter()); + ColumnFamily cf = table.getColumnFamilyStore(LOCATION_CF).getColumnFamily(new NamesQueryFilter(LOCATION_KEY, LOCATION_KEY, TOKEN)); long oldTokenColumnTimestamp = cf.getColumn(SystemTable.TOKEN).timestamp(); /* create the "Token" whose value is the new token. */ IColumn tokenColumn = new Column(SystemTable.TOKEN, p.getTokenFactory().toByteArray(token), oldTokenColumnTimestamp + 1); @@ -71,7 +73,7 @@ public class SystemTable { /* Read the system table to retrieve the storage ID and the generation */ Table table = Table.open(Table.SYSTEM_TABLE); - ColumnFamily cf = table.getColumnFamilyStore(LOCATION_CF).getColumnFamily(LOCATION_KEY, LOCATION_KEY + ":" + GENERATION, new IdentityFilter()); + ColumnFamily cf = table.getColumnFamilyStore(LOCATION_CF).getColumnFamily(new NamesQueryFilter(LOCATION_KEY, LOCATION_KEY, GENERATION)); IPartitioner p = StorageService.getPartitioner(); if (cf == null) diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 3072adfaec..dada97f0e0 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -43,10 +43,7 @@ import org.apache.cassandra.net.io.IStreamComplete; import org.apache.cassandra.net.io.StreamContextManager; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.*; -import org.apache.cassandra.db.filter.QueryFilter; -import org.apache.cassandra.db.filter.SliceQueryFilter; -import org.apache.cassandra.db.filter.NamesQueryFilter; -import org.apache.cassandra.db.filter.TimeQueryFilter; +import org.apache.cassandra.db.filter.*; import org.apache.log4j.Logger; @@ -502,23 +499,16 @@ public class Table */ @Deprecated // CF should be our atom of work, not Row public Row get(String key) throws IOException - { + { Row row = new Row(table_, key); - Set columnFamilies = tableMetadata_.getColumnFamilies(); - long start = System.currentTimeMillis(); - for ( String columnFamily : columnFamilies ) + for (String columnFamily : getColumnFamilies()) { - ColumnFamilyStore cfStore = columnFamilyStores_.get(columnFamily); - if ( cfStore != null ) - { - ColumnFamily cf = cfStore.getColumnFamily(key, columnFamily, new IdentityFilter()); - if ( cf != null ) - row.addColumnFamily(cf); + ColumnFamily cf = get(key, columnFamily); + if (cf != null) + { + row.addColumnFamily(cf); } } - - long timeTaken = System.currentTimeMillis() - start; - dbAnalyticsSource_.updateReadStatistics(timeTaken); return row; } @@ -527,25 +517,22 @@ public class Table * Selects the specified column family for the specified key. */ @Deprecated // single CFs could be larger than memory - public ColumnFamily get(String key, String columnFamilyColumn) throws IOException + public ColumnFamily get(String key, String cfName) throws IOException { - String[] values = RowMutation.getColumnAndColumnFamily(columnFamilyColumn); - long start = System.currentTimeMillis(); + assert !cfName.contains(":") : cfName; + String[] values = RowMutation.getColumnAndColumnFamily(cfName); ColumnFamilyStore cfStore = columnFamilyStores_.get(values[0]); - assert cfStore != null : "Column family " + columnFamilyColumn + " has not been defined"; - ColumnFamily columnFamily = cfStore.getColumnFamily(key, columnFamilyColumn, new IdentityFilter()); - long timeTaken = System.currentTimeMillis() - start; - dbAnalyticsSource_.updateReadStatistics(timeTaken); - return columnFamily; + assert cfStore != null : "Column family " + cfName + " has not been defined"; + return cfStore.getColumnFamily(new IdentityQueryFilter(key, cfName)); } /** * Selects only the specified column family for the specified key. */ - public Row getRow(String key, String cf) throws IOException + public Row getRow(String key, String cfName) throws IOException { Row row = new Row(table_, key); - ColumnFamily columnFamily = get(key, cf); + ColumnFamily columnFamily = get(key, cfName); if ( columnFamily != null ) row.addColumnFamily(columnFamily); return row; @@ -581,28 +568,14 @@ public class Table return getRow(key, filter); } - private Row getRow(String key, QueryFilter filter) throws IOException + public Row getRow(String key, QueryFilter filter) throws IOException { ColumnFamilyStore cfStore = columnFamilyStores_.get(filter.getColumnFamilyName()); Row row = new Row(table_, key); - long start1 = System.currentTimeMillis(); - try - { - ColumnFamily columnFamily = cfStore.getColumnFamily(filter); - if (columnFamily != null) - row.addColumnFamily(columnFamily); - long timeTaken = System.currentTimeMillis() - start1; - dbAnalyticsSource_.updateReadStatistics(timeTaken); - return row; - } - catch (InterruptedException e) - { - throw new RuntimeException(e); - } - catch (ExecutionException e) - { - throw new RuntimeException(e); - } + ColumnFamily columnFamily = cfStore.getColumnFamily(filter); + if (columnFamily != null) + row.addColumnFamily(columnFamily); + return row; } /** @@ -612,8 +585,6 @@ public class Table */ void apply(Row row) throws IOException { - /* Add row to the commit log. */ - long start = System.currentTimeMillis(); CommitLog.CommitLogContext cLogCtx = CommitLog.open().add(row); for (ColumnFamily columnFamily : row.getColumnFamilies()) @@ -621,9 +592,6 @@ public class Table ColumnFamilyStore cfStore = columnFamilyStores_.get(columnFamily.name()); cfStore.apply(row.key(), columnFamily, cLogCtx); } - - long timeTaken = System.currentTimeMillis() - start; - dbAnalyticsSource_.updateWriteStatistics(timeTaken); } void applyNow(Row row) throws IOException @@ -772,7 +740,7 @@ public class Table } // make sure there is actually non-tombstone content associated w/ this key // TODO record the key source(s) somehow and only check that source (e.g., memtable or sstable) - if (cfs.getColumnFamily(current, columnFamily, new IdentityFilter(), Integer.MAX_VALUE) != null) + if (cfs.getColumnFamily(new SliceQueryFilter(current, columnFamily, "", "", true, 0, 1)) != null) { keys.add(current); } diff --git a/src/java/org/apache/cassandra/db/filter/IdentityQueryFilter.java b/src/java/org/apache/cassandra/db/filter/IdentityQueryFilter.java new file mode 100644 index 0000000000..c00fe52fdf --- /dev/null +++ b/src/java/org/apache/cassandra/db/filter/IdentityQueryFilter.java @@ -0,0 +1,20 @@ +package org.apache.cassandra.db.filter; + +import org.apache.cassandra.db.SuperColumn; + +public class IdentityQueryFilter extends SliceQueryFilter +{ + /** + * Only for use in testing; will read entire CF into memory. + */ + public IdentityQueryFilter(String key, String columnFamilyColumn) + { + super(key, columnFamilyColumn, "", "", true, 0, Integer.MAX_VALUE); + } + + @Override + public void filterSuperColumn(SuperColumn superColumn) + { + // no filtering done, deliberately + } +} diff --git a/src/java/org/apache/cassandra/io/SequenceFile.java b/src/java/org/apache/cassandra/io/SequenceFile.java index db5b35f26e..f4ec2db6db 100644 --- a/src/java/org/apache/cassandra/io/SequenceFile.java +++ b/src/java/org/apache/cassandra/io/SequenceFile.java @@ -829,74 +829,4 @@ public class SequenceFile { return new BufferReader(filename, size); } - - /** - * Efficiently writes a UTF8 string to the buffer. - * Assuming all Strings that are passed in have length - * that can be represented as a short i.e length of the - * string is <= 65535 - * - * @param buffer buffer to write the serialize version into - * @param str string to serialize - */ - protected static void writeUTF(ByteBuffer buffer, String str) - { - int strlen = str.length(); - int utflen = 0; - int c, count = 0; - - /* use charAt instead of copying String to char array */ - for (int i = 0; i < strlen; i++) - { - c = str.charAt(i); - if ((c >= 0x0001) && (c <= 0x007F)) - { - utflen++; - } - else if (c > 0x07FF) - { - utflen += 3; - } - else - { - utflen += 2; - } - } - - byte[] bytearr = new byte[utflen + 2]; - bytearr[count++] = (byte) ((utflen >>> 8) & 0xFF); - bytearr[count++] = (byte) ((utflen >>> 0) & 0xFF); - - int i = 0; - for (i = 0; i < strlen; i++) - { - c = str.charAt(i); - if (!((c >= 0x0001) && (c <= 0x007F))) - break; - bytearr[count++] = (byte) c; - } - - for (; i < strlen; i++) - { - c = str.charAt(i); - if ((c >= 0x0001) && (c <= 0x007F)) - { - bytearr[count++] = (byte) c; - - } - else if (c > 0x07FF) - { - bytearr[count++] = (byte) (0xE0 | ((c >> 12) & 0x0F)); - bytearr[count++] = (byte) (0x80 | ((c >> 6) & 0x3F)); - bytearr[count++] = (byte) (0x80 | ((c >> 0) & 0x3F)); - } - else - { - bytearr[count++] = (byte) (0xC0 | ((c >> 6) & 0x1F)); - bytearr[count++] = (byte) (0x80 | ((c >> 0) & 0x3F)); - } - } - buffer.put(bytearr, 0, utflen + 2); - } - } diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java index a102679d10..ec41296b12 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java @@ -30,6 +30,7 @@ import org.junit.Test; import static junit.framework.Assert.assertEquals; import org.apache.cassandra.CleanupHelper; +import org.apache.cassandra.db.filter.IdentityQueryFilter; import org.apache.cassandra.io.SSTableReader; public class ColumnFamilyStoreTest extends CleanupHelper @@ -115,7 +116,7 @@ public class ColumnFamilyStoreTest extends CleanupHelper List ssTables = table.getAllSSTablesOnDisk(); assertEquals(1, ssTables.size()); ssTables.get(0).forceBloomFilterFailures(); - ColumnFamily cf = store.getColumnFamily("key2", "Standard1:Column1", new IdentityFilter()); + ColumnFamily cf = store.getColumnFamily(new IdentityQueryFilter("key2", "Standard1:Column1")); assertNull(cf); } } diff --git a/test/unit/org/apache/cassandra/db/RemoveColumnFamilyTest.java b/test/unit/org/apache/cassandra/db/RemoveColumnFamilyTest.java index 9b817a6ddd..6bf2131ee1 100644 --- a/test/unit/org/apache/cassandra/db/RemoveColumnFamilyTest.java +++ b/test/unit/org/apache/cassandra/db/RemoveColumnFamilyTest.java @@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException; import org.junit.Test; import static junit.framework.Assert.assertNull; +import org.apache.cassandra.db.filter.IdentityQueryFilter; public class RemoveColumnFamilyTest { @@ -44,7 +45,7 @@ public class RemoveColumnFamilyTest rm.delete("Standard1", 1); rm.apply(); - ColumnFamily retrieved = store.getColumnFamily("key1", "Standard1:Column1", new IdentityFilter()); + ColumnFamily retrieved = store.getColumnFamily(new IdentityQueryFilter("key1", "Standard1:Column1")); assert retrieved.isMarkedForDelete(); assertNull(retrieved.getColumn("Column1")); assertNull(ColumnFamilyStore.removeDeleted(retrieved, Integer.MAX_VALUE)); diff --git a/test/unit/org/apache/cassandra/db/RemoveColumnFamilyWithFlush1Test.java b/test/unit/org/apache/cassandra/db/RemoveColumnFamilyWithFlush1Test.java index f65109482e..de2792db68 100644 --- a/test/unit/org/apache/cassandra/db/RemoveColumnFamilyWithFlush1Test.java +++ b/test/unit/org/apache/cassandra/db/RemoveColumnFamilyWithFlush1Test.java @@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException; import org.junit.Test; import static junit.framework.Assert.assertNull; +import org.apache.cassandra.db.filter.IdentityQueryFilter; public class RemoveColumnFamilyWithFlush1Test { @@ -46,7 +47,7 @@ public class RemoveColumnFamilyWithFlush1Test rm.delete("Standard1", 1); rm.apply(); - ColumnFamily retrieved = store.getColumnFamily("key1", "Standard1", new IdentityFilter()); + ColumnFamily retrieved = store.getColumnFamily(new IdentityQueryFilter("key1", "Standard1")); assert retrieved.isMarkedForDelete(); assertNull(retrieved.getColumn("Column1")); assertNull(ColumnFamilyStore.removeDeleted(retrieved, Integer.MAX_VALUE)); diff --git a/test/unit/org/apache/cassandra/db/RemoveColumnFamilyWithFlush2Test.java b/test/unit/org/apache/cassandra/db/RemoveColumnFamilyWithFlush2Test.java index 2d424ba38f..e8b6d1ac6c 100644 --- a/test/unit/org/apache/cassandra/db/RemoveColumnFamilyWithFlush2Test.java +++ b/test/unit/org/apache/cassandra/db/RemoveColumnFamilyWithFlush2Test.java @@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException; import org.junit.Test; import static junit.framework.Assert.assertNull; +import org.apache.cassandra.db.filter.IdentityQueryFilter; public class RemoveColumnFamilyWithFlush2Test { @@ -44,7 +45,7 @@ public class RemoveColumnFamilyWithFlush2Test rm.apply(); store.forceBlockingFlush(); - ColumnFamily retrieved = store.getColumnFamily("key1", "Standard1:Column1", new IdentityFilter()); + ColumnFamily retrieved = store.getColumnFamily(new IdentityQueryFilter("key1", "Standard1:Column1")); assert retrieved.isMarkedForDelete(); assertNull(retrieved.getColumn("Column1")); assertNull(ColumnFamilyStore.removeDeleted(retrieved, Integer.MAX_VALUE)); diff --git a/test/unit/org/apache/cassandra/db/RemoveColumnTest.java b/test/unit/org/apache/cassandra/db/RemoveColumnTest.java index 82eeb5a504..156cc687ca 100644 --- a/test/unit/org/apache/cassandra/db/RemoveColumnTest.java +++ b/test/unit/org/apache/cassandra/db/RemoveColumnTest.java @@ -24,6 +24,8 @@ import java.util.concurrent.ExecutionException; import org.junit.Test; import static junit.framework.Assert.assertNull; +import org.apache.cassandra.db.filter.IdentityQueryFilter; +import org.apache.cassandra.db.filter.NamesQueryFilter; public class RemoveColumnTest { @@ -45,8 +47,9 @@ public class RemoveColumnTest rm.delete("Standard1:Column1", 1); rm.apply(); - ColumnFamily retrieved = store.getColumnFamily("key1", "Standard1", new IdentityFilter()); + ColumnFamily retrieved = store.getColumnFamily(new NamesQueryFilter("key1", "Standard1", "Column1")); assert retrieved.getColumn("Column1").isMarkedForDelete(); assertNull(ColumnFamilyStore.removeDeleted(retrieved, Integer.MAX_VALUE)); + assertNull(ColumnFamilyStore.removeDeleted(store.getColumnFamily(new IdentityQueryFilter("key1", "Standard1")), Integer.MAX_VALUE)); } } \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/db/RemoveSubColumnTest.java b/test/unit/org/apache/cassandra/db/RemoveSubColumnTest.java index e91ce619c9..ff95be9bef 100644 --- a/test/unit/org/apache/cassandra/db/RemoveSubColumnTest.java +++ b/test/unit/org/apache/cassandra/db/RemoveSubColumnTest.java @@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException; import org.junit.Test; import static junit.framework.Assert.assertNull; +import org.apache.cassandra.db.filter.IdentityQueryFilter; public class RemoveSubColumnTest { @@ -45,7 +46,7 @@ public class RemoveSubColumnTest rm.delete("Super1:SC1:Column1", 1); rm.apply(); - ColumnFamily retrieved = store.getColumnFamily("key1", "Super1:SC1", new IdentityFilter()); + ColumnFamily retrieved = store.getColumnFamily(new IdentityQueryFilter("key1", "Super1:SC1")); assert retrieved.getColumn("SC1").getSubColumn("Column1").isMarkedForDelete(); assertNull(ColumnFamilyStore.removeDeleted(retrieved, Integer.MAX_VALUE)); } diff --git a/test/unit/org/apache/cassandra/db/RemoveSuperColumnTest.java b/test/unit/org/apache/cassandra/db/RemoveSuperColumnTest.java index 42bee3b176..6bc10e75f4 100644 --- a/test/unit/org/apache/cassandra/db/RemoveSuperColumnTest.java +++ b/test/unit/org/apache/cassandra/db/RemoveSuperColumnTest.java @@ -24,12 +24,13 @@ import java.util.concurrent.Future; import java.util.List; import java.util.Collection; import java.util.Arrays; +import java.util.TreeSet; -import org.apache.commons.lang.StringUtils; import org.junit.Test; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertEquals; +import org.apache.cassandra.db.filter.IdentityQueryFilter; import org.apache.cassandra.db.filter.NamesQueryFilter; public class RemoveSuperColumnTest @@ -65,24 +66,17 @@ public class RemoveSuperColumnTest private void validateRemoveTwoSources() throws IOException { ColumnFamilyStore store = Table.open("Table1").getColumnFamilyStore("Super1"); - List families = store.getColumnFamilies("key1", "Super1", new IdentityFilter()); - assert families.size() == 2 : StringUtils.join(families, ", "); - assert families.get(0).getAllColumns().first().getMarkedForDeleteAt() == 1; // delete marker, just added - assert !families.get(1).getAllColumns().first().isMarkedForDelete(); // flushed old version - ColumnFamily resolved = ColumnFamily.resolve(families); + ColumnFamily resolved = store.getColumnFamily(new NamesQueryFilter("key1", "Super1", "SC1")); assert resolved.getAllColumns().first().getMarkedForDeleteAt() == 1; - Collection subColumns = resolved.getAllColumns().first().getSubColumns(); - assert subColumns.size() == 1; - assert subColumns.iterator().next().timestamp() == 0; + assert resolved.getAllColumns().first().getSubColumns().size() == 0; assertNull(ColumnFamilyStore.removeDeleted(resolved, Integer.MAX_VALUE)); + assertNull(ColumnFamilyStore.removeDeleted(store.getColumnFamily(new IdentityQueryFilter("key1", "Super1")), Integer.MAX_VALUE)); } private void validateRemoveCompacted() throws IOException { ColumnFamilyStore store = Table.open("Table1").getColumnFamilyStore("Super1"); - List families = store.getColumnFamilies("key1", "Super1", new IdentityFilter()); - assert families.size() == 1 : StringUtils.join(families, ", "); - ColumnFamily resolved = families.get(0); + ColumnFamily resolved = store.getColumnFamily(new NamesQueryFilter("key1", "Super1", "SC1")); assert resolved.getAllColumns().first().getMarkedForDeleteAt() == 1; Collection subColumns = resolved.getAllColumns().first().getSubColumns(); assert subColumns.size() == 0; @@ -124,10 +118,7 @@ public class RemoveSuperColumnTest private void validateRemoveWithNewData() throws IOException { ColumnFamilyStore store = Table.open("Table1").getColumnFamilyStore("Super2"); - List families; - ColumnFamily resolved; - - resolved = store.getColumnFamily(new NamesQueryFilter("key1", "Super2:SC1", "Column2")); + ColumnFamily resolved = store.getColumnFamily(new NamesQueryFilter("key1", "Super2:SC1", "Column2")); validateNewDataFamily(resolved); } diff --git a/test/unit/org/apache/cassandra/db/TableTest.java b/test/unit/org/apache/cassandra/db/TableTest.java index 9de261ce66..5a34a8286a 100644 --- a/test/unit/org/apache/cassandra/db/TableTest.java +++ b/test/unit/org/apache/cassandra/db/TableTest.java @@ -26,6 +26,7 @@ import org.junit.Test; import static junit.framework.Assert.*; import org.apache.cassandra.CleanupHelper; +import org.apache.cassandra.db.filter.NamesQueryFilter; import org.apache.cassandra.io.SSTableReader; public class TableTest extends CleanupHelper @@ -71,10 +72,10 @@ public class TableTest extends CleanupHelper { Row result; - result = table.getRow(TEST_KEY, "Standard1:col1"); + result = table.getRow(TEST_KEY, new NamesQueryFilter(TEST_KEY, "Standard1", "col1")); assertColumns(result.getColumnFamily("Standard1"), "col1"); - result = table.getRow(TEST_KEY, "Standard1:col3"); + result = table.getRow(TEST_KEY, new NamesQueryFilter(TEST_KEY, "Standard1", "col3")); assertColumns(result.getColumnFamily("Standard1"), "col3"); } }; @@ -367,8 +368,7 @@ public class TableTest extends CleanupHelper public static void assertColumns(ColumnFamily columnFamily, String... columnNames) { - assertNotNull(columnFamily); - SortedSet columns = columnFamily.getAllColumns(); + SortedSet columns = columnFamily == null ? new TreeSet() : columnFamily.getAllColumns(); List L = new ArrayList(); for (IColumn column : columns) {