Index: src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestDeleteAll.java =================================================================== --- src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestDeleteAll.java (revision 0) +++ src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestDeleteAll.java (revision 0) @@ -0,0 +1,155 @@ +/** + * Copyright 2007 The Apache Software Foundation + * + * 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.hadoop.hbase; + +import java.io.IOException; +import java.util.Map; +import java.util.TreeMap; + +import org.apache.hadoop.dfs.MiniDFSCluster; +import org.apache.hadoop.hbase.filter.StopRowFilter; +import org.apache.hadoop.hbase.filter.WhileMatchRowFilter; +import org.apache.hadoop.io.Text; +import org.apache.commons.logging.*; + +import java.util.List; + +/** + * Test the functionality of deleteAll. + */ +public class TestDeleteAll extends HBaseTestCase { + static final Log LOG = LogFactory.getLog(TestDeleteAll.class); + private MiniDFSCluster miniHdfs; + + @Override + protected void setUp() throws Exception { + super.setUp(); + this.miniHdfs = new MiniDFSCluster(this.conf, 1, true, null); + } + + /** + * Tests for HADOOP-1550. + * @throws Exception + */ + public void testDeleteAll() throws Exception { + HRegion region = null; + HRegionIncommon region_incommon = null; + HLog hlog = new HLog(this.miniHdfs.getFileSystem(), this.testDir, + this.conf, null); + + try{ + HTableDescriptor htd = createTableDescriptor(getName()); + HRegionInfo hri = new HRegionInfo(htd, null, null); + region = new HRegion(this.testDir, hlog, this.miniHdfs.getFileSystem(), + this.conf, hri, null, null); + region_incommon = new HRegionIncommon(region); + + makeSureItWorks(region, region_incommon, false); + LOG.info("+++++++++++++++++++++++++++++++++++++++"); + makeSureItWorks(region, region_incommon, true); + + } finally { + if (region != null) { + try { + region.close(); + } catch (Exception e) { + e.printStackTrace(); + } + } + hlog.closeAndDelete(); + } + } + + private void makeSureItWorks(HRegion region, HRegionIncommon region_incommon, + boolean flush) + throws Exception{ + // insert a few versions worth of data for a row + Text row = new Text("test_row"); + long t0 = System.currentTimeMillis(); + long t1 = t0 - 15000; + long t2 = t1 - 15000; + + Text colA = new Text(COLUMNS[0].toString() + "a"); + Text colB = new Text(COLUMNS[0].toString() + "b"); + Text colC = new Text(COLUMNS[0].toString() + "c"); + + long lock = region_incommon.startUpdate(row); + region_incommon.put(lock, colA, cellData(0, flush).getBytes()); + region_incommon.put(lock, colB, cellData(0, flush).getBytes()); + region_incommon.put(lock, colC, cellData(0, flush).getBytes()); + region_incommon.commit(lock, t0); + + lock = region_incommon.startUpdate(row); + region_incommon.put(lock, colA, cellData(1, flush).getBytes()); + region_incommon.put(lock, colB, cellData(1, flush).getBytes()); + region_incommon.put(lock, colC, cellData(1, flush).getBytes()); + region_incommon.commit(lock, t1); + + lock = region_incommon.startUpdate(row); + region_incommon.put(lock, colA, cellData(2, flush).getBytes()); + region_incommon.put(lock, colB, cellData(2, flush).getBytes()); + region_incommon.put(lock, colC, cellData(2, flush).getBytes()); + region_incommon.commit(lock, t2); + + if (flush) {region_incommon.flushcache();} + + // call delete all at a timestamp, make sure only the most recent stuff is left behind + region.deleteAll(row, t1); + if (flush) {region_incommon.flushcache();} + assertCellValueEquals(region, row, colA, t0, cellData(0, flush)); + assertCellValueEquals(region, row, colA, t1, null); + assertCellValueEquals(region, row, colA, t2, null); + + // call delete all w/o a timestamp, make sure nothing is left. + region.deleteAll(row, HConstants.LATEST_TIMESTAMP); + if (flush) {region_incommon.flushcache();} + assertCellValueEquals(region, row, colA, t0, null); + assertCellValueEquals(region, row, colA, t1, null); + assertCellValueEquals(region, row, colA, t2, null); + + } + + private void assertCellValueEquals(final HRegion region, final Text row, + final Text column, final long timestamp, final String value) + throws IOException { + Map result = region.getFull(row, timestamp); + byte[] cell_value = result.get(column); + if(value == null){ + assertEquals(column.toString() + " at timestamp " + timestamp, null, cell_value); + } else { + if (cell_value == null) { + fail(column.toString() + " at timestamp " + timestamp + "\" was expected to be \"" + value + " but was null"); + } + assertEquals(column.toString() + " at timestamp " + timestamp, value, new String(cell_value)); + } + + } + + private String cellData(int tsNum, boolean flush){ + return "t" + tsNum + " data" + (flush ? " - with flush" : ""); + } + + protected void tearDown() throws Exception { + if (this.miniHdfs != null) { + this.miniHdfs.shutdown(); + } + super.tearDown(); + } +} Index: src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStore.java =================================================================== --- src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStore.java (revision 602176) +++ src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStore.java (working copy) @@ -268,7 +268,7 @@ (versions - results.size()))); } return results; - + } finally { this.lock.readLock().unlock(); } @@ -289,9 +289,28 @@ SortedMap tailMap = map.tailMap(origin); for (Map.Entry es: tailMap.entrySet()) { HStoreKey key = es.getKey(); - if (!key.matchesRowCol(origin)) { - break; + + // if there's no column name, then compare rows and timestamps + if (origin.getColumn().toString().equals("")) { + // if the current and origin row don't match, then we can jump + // out of the loop entirely. + if (!key.getRow().equals(origin.getRow())) { + break; + } + // if the rows match but the timestamp is newer, skip it so we can + // get to the ones we actually want. + if (key.getTimestamp() > origin.getTimestamp()) { + continue; + } } + else{ // compare rows and columns + // if the key doesn't match the row and column, then we're done, since + // all the cells are ordered. + if (!key.matchesRowCol(origin)) { + break; + } + } + if (!HLogEdit.isDeleted(es.getValue())) { result.add(key); if (versions != HConstants.ALL_VERSIONS && result.size() >= versions) { @@ -1429,6 +1448,7 @@ */ void getFull(HStoreKey key, TreeMap results) throws IOException { + Map> deletes = new HashMap>(); this.lock.readLock().lock(); memcache.getFull(key, results); @@ -1447,7 +1467,7 @@ Text readcol = readkey.getColumn(); if (results.get(readcol) == null && key.matchesWithoutColumn(readkey)) { - if(HLogEdit.isDeleted(readval.get())) { + if(isDeleted(readkey, readval.get(), true, deletes)) { break; } results.put(new Text(readcol), readval.get()); @@ -1598,6 +1618,8 @@ MapFile.Reader map = maparray[i]; synchronized(map) { map.reset(); + + // do the priming read ImmutableBytesWritable readval = new ImmutableBytesWritable(); HStoreKey readkey = (HStoreKey)map.getClosest(origin, readval); if (readkey == null) { @@ -1607,33 +1629,82 @@ // BEFORE. continue; } - if (!readkey.matchesRowCol(origin)) { - continue; - } - if (!isDeleted(readkey, readval.get(), false, null) && - !keys.contains(readkey)) { - keys.add(new HStoreKey(readkey)); - } - for (readval = new ImmutableBytesWritable(); - map.next(readkey, readval) && - readkey.matchesRowCol(origin); - readval = new ImmutableBytesWritable()) { - if (!isDeleted(readkey, readval.get(), false, null) && - !keys.contains(readkey)) { - keys.add(new HStoreKey(readkey)); - if (versions != ALL_VERSIONS && keys.size() >= versions) { - break; + + do{ + // if the row matches, we might want this one. + if(rowMatches(origin, readkey)){ + // if the cell matches, then we definitely want this key. + if (cellMatches(origin, readkey)) { + // store the key if it isn't deleted or superceeded by what's + // in the memcache + if (!isDeleted(readkey, readval.get(), false, null) && + !keys.contains(readkey)) { + keys.add(new HStoreKey(readkey)); + + // if we've collected enough versions, then exit the loop. + if (versions != ALL_VERSIONS && keys.size() >= versions) { + break; + } + } + } else { + // the cell doesn't match, but there might be more with different + // timestamps, so move to the next key + continue; } + } else{ + // the row doesn't match, so we've gone too far. + break; } - } + }while(map.next(readkey, readval)); // advance to the next key } } + return keys; } finally { this.lock.readLock().unlock(); } } + /** + * Test that the @param target matches the @param origin. If the @param + * origin has an empty column, then it's assumed to mean any column + * matches and only match on row and timestamp. Otherwise, it compares the + * keys with HStoreKey.matchesRowCol(). + * @param origin The key we're testing against + * @param target The key we're testing + */ + private boolean cellMatches(HStoreKey origin, HStoreKey target){ + // if the origin's column is empty, then we're matching any column + if (origin.getColumn().equals(new Text())){ + // if the row matches, then... + if (target.getRow().equals(origin.getRow())) { + // check the timestamp + return target.getTimestamp() <= origin.getTimestamp(); + } else { + return false; + } + } else { // otherwise, we want to match on row and column + return target.matchesRowCol(origin); + } + } + + /** + * Test that the @param target matches the @param origin. If the @param origin + * has an empty column, then it just tests row equivalence. Otherwise, it uses + * HStoreKey.matchesRowCol(). + * @param origin Key we're testing against + * @param target Key we're testing + */ + private boolean rowMatches(HStoreKey origin, HStoreKey target){ + // if the origin's column is empty, then we're matching any column + if (origin.getColumn().equals(new Text())){ + // if the row matches, then... + return target.getRow().equals(origin.getRow()); + } else { // otherwise, we want to match on row and column + return target.matchesRowCol(origin); + } + } + /* * Data structure to hold result of a look at store file sizes. */ Index: src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java =================================================================== --- src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java (revision 602176) +++ src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java (working copy) @@ -1444,6 +1444,14 @@ region.deleteAll(row, column, timestamp); } + /** {@inheritDoc} */ + public void deleteAll(final Text regionName, final Text row, + final long timestamp) + throws IOException { + HRegion region = getRegion(regionName); + region.deleteAll(row, timestamp); + } + /** * @return Info on this server. */ Index: src/contrib/hbase/src/java/org/apache/hadoop/hbase/HTable.java =================================================================== --- src/contrib/hbase/src/java/org/apache/hadoop/hbase/HTable.java (revision 602176) +++ src/contrib/hbase/src/java/org/apache/hadoop/hbase/HTable.java (working copy) @@ -742,6 +742,53 @@ } } + /** + * Completely delete the row's cells of the same timestamp or older. + * + * @param row Key of the row you want to completely delete. + * @param ts Timestamp of cells to delete + */ + public void deleteAll(final Text row, long ts) + throws IOException { + checkClosed(); + for(int tries = 0; tries < numRetries; tries++) { + HRegionLocation r = getRegionLocation(row); + HRegionInterface server = + connection.getHRegionConnection(r.getServerAddress()); + try { + server.deleteAll(r.getRegionInfo().getRegionName(), row, ts); + break; + + } catch (IOException e) { + if (e instanceof RemoteException) { + e = RemoteExceptionHandler.decodeRemoteException((RemoteException) e); + } + if (tries == numRetries - 1) { + throw e; + } + if (LOG.isDebugEnabled()) { + LOG.debug("reloading table servers because: " + e.getMessage()); + } + tableServers = connection.reloadTableServers(tableName); + } + try { + Thread.sleep(this.pause); + } catch (InterruptedException x) { + // continue + } + } + } + + /** + * Completely delete the row's cells. + * + * @param row Key of the row you want to completely delete. + */ + public void deleteAll(final Text row) + throws IOException { + deleteAll(row, HConstants.LATEST_TIMESTAMP); + } + /** * Abort a row mutation. * Index: src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInterface.java =================================================================== --- src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInterface.java (revision 602176) +++ src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInterface.java (working copy) @@ -134,6 +134,18 @@ */ public void deleteAll(Text regionName, Text row, Text column, long timestamp) throws IOException; + + /** + * Delete all cells that match the passed row and whose + * timestamp is equal-to or older than the passed timestamp. + * + * @param regionName region name + * @param row row key + * @param timestamp Delete all entries that have this timestamp or older + * @throws IOException + */ + public void deleteAll(Text regionName, Text row, long timestamp) + throws IOException; // // remote scanner interface Index: src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java =================================================================== --- src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java (revision 602176) +++ src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java (working copy) @@ -1060,7 +1060,7 @@ */ private List getKeys(final HStoreKey origin, final int versions) throws IOException { - + List keys = null; Text colFamily = HStoreKey.extractFamily(origin.getColumn()); HStore targetStore = stores.get(colFamily); @@ -1071,7 +1071,7 @@ } return keys; } - + /** * Return an iterator that scans over the HRegion, returning the indicated * columns for only the rows that match the data filter. This Iterator must @@ -1248,6 +1248,33 @@ releaseRowLock(row); } } + + /** + * Delete all cells of the same age as the passed timestamp or older. + * @param row + * @param ts Delete all entries that have this timestamp or older + * @throws IOException + */ + public void deleteAll(final Text row, final long ts) + throws IOException { + + obtainRowLock(row); + + try { + for(Map.Entry store : stores.entrySet()){ + List keys = store.getValue().getKeys(new HStoreKey(row, ts), ALL_VERSIONS); + + TreeMap edits = new TreeMap(); + for (HStoreKey key: keys) { + edits.put(key, HLogEdit.deleteBytes.get()); + } + update(edits); + } + } finally { + releaseRowLock(row); + } + } + /** * Delete one or many cells.