Changeset 4148

Timestamp:

2010-07-16 12:18:47 (3 years ago)

Author:

evert

Message:

Delete of a record now puts a deleted-flag and avoids that Rowlog information for that record is deleted as well.
A cleanup process will be needed for any records that are flaged as deleted and don't have any remaining rowlog tasks to perform.

Location:

projects/lily/trunk

Files:

• repository/impl/src/main/java/org/lilycms/repository/impl/HBaseRepository.java (modified) (10 diffs)
• repository/impl/src/test/java/org/lilycms/repository/impl/test/AbstractRepositoryTest.java (modified) (3 diffs)
• repository/impl/src/test/java/org/lilycms/repository/impl/test/AvroRepositoryTest.java (modified) (1 diff)
• repository/impl/src/test/java/org/lilycms/repository/impl/test/HBaseRepositoryTest.java (modified) (1 diff)
• testfw/src/main/java/org/lilycms/testfw/HBaseProxy.java (modified) (2 diffs)

projects/lily/trunk/repository/impl/src/main/java/org/lilycms/repository/impl/HBaseRepository.java

@@ -84 +84 @@
     private static final byte[] CURRENT_VERSION_COLUMN_NAME = Bytes.toBytes("$CurrentVersion");
     private static final byte[] LOCK_COLUMN_NAME = Bytes.toBytes("$Lock");
+    private static final byte[] DELETED_COLUMN_NAME = Bytes.toBytes("$Deleted");
     private static final byte[] NON_VERSIONED_RECORDTYPEID_COLUMN_NAME = Bytes.toBytes("$NonVersionableRecordTypeId");
     private static final byte[] NON_VERSIONED_RECORDTYPEVERSION_COLUMN_NAME = Bytes
@@ -212 +213 @@
             checkCreatePreconditions(record);
 
-            // Creating an existing record is not allowed. Use update instead.
-            if (recordTable.exists(new Get(rowId, rowLock))) {
-                throw new RecordExistsException(newRecord);
-            }
-
+            // Check if a previous incarnation of the record existed and clear data if needed
+            reincarnateRecord(newRecord, rowId, rowLock);
+            
             Record dummyOriginalRecord = newRecord();
             Put put = new Put(newRecord.getId().toBytes(), rowLock);
+            put.add(systemColumnFamilies.get(Scope.NON_VERSIONED), DELETED_COLUMN_NAME, Bytes.toBytes(false));
             RecordEvent recordEvent = new RecordEvent();
             recordEvent.setType(Type.CREATE);
@@ -272 +272 @@
     }
 
+    private void reincarnateRecord(Record newRecord, byte[] rowId, RowLock rowLock) throws IOException,
+            RecordExistsException {
+        Get get = new Get(rowId, rowLock);
+        if (recordTable.exists(get)) {
+            get.addColumn(systemColumnFamilies.get(Scope.NON_VERSIONED), DELETED_COLUMN_NAME);
+            get.addColumn(systemColumnFamilies.get(Scope.NON_VERSIONED), CURRENT_VERSION_COLUMN_NAME);
+            Result result = recordTable.get(get);
+            if (result == null || result.isEmpty()) {
+                throw new RecordExistsException(newRecord);
+            } else {
+                byte[] deleted = result.getValue(systemColumnFamilies.get(Scope.NON_VERSIONED), DELETED_COLUMN_NAME);
+                if ((deleted != null) && (Bytes.toBoolean(deleted))) {
+                    if (null != result.getValue(systemColumnFamilies.get(Scope.NON_VERSIONED), CURRENT_VERSION_COLUMN_NAME)) {
+                        clearData(rowId, rowLock);
+                    }
+                } else {
+                    throw new RecordExistsException(newRecord);
+                }
+            }
+        }
+    }
+
+    private void clearData(byte[] rowId, RowLock rowLock) throws IOException {
+        Delete delete = new Delete(rowId, -1, rowLock);
+        delete.deleteFamily(columnFamilies.get(Scope.NON_VERSIONED));
+        delete.deleteFamily(columnFamilies.get(Scope.VERSIONED));
+        delete.deleteFamily(columnFamilies.get(Scope.VERSIONED_MUTABLE));
+        delete.deleteColumns(systemColumnFamilies.get(Scope.NON_VERSIONED), CURRENT_VERSION_COLUMN_NAME);
+        delete.deleteColumns(systemColumnFamilies.get(Scope.NON_VERSIONED), NON_VERSIONED_RECORDTYPEID_COLUMN_NAME);
+        delete.deleteColumns(systemColumnFamilies.get(Scope.NON_VERSIONED), NON_VERSIONED_RECORDTYPEVERSION_COLUMN_NAME);
+        delete.deleteColumns(systemColumnFamilies.get(Scope.VERSIONED), VERSIONED_RECORDTYPEID_COLUMN_NAME);
+        delete.deleteColumns(systemColumnFamilies.get(Scope.VERSIONED), VERSIONED_RECORDTYPEVERSION_COLUMN_NAME);
+        delete.deleteColumns(systemColumnFamilies.get(Scope.VERSIONED), VERSIONED_MUTABLE_RECORDTYPEID_COLUMN_NAME);
+        delete.deleteColumns(systemColumnFamilies.get(Scope.VERSIONED), VERSIONED_MUTABLE_RECORDTYPEVERSION_COLUMN_NAME);
+        recordTable.delete(delete);
+    }
+
     private void checkCreatePreconditions(Record record) throws InvalidRecordException {
         ArgumentValidator.notNull(record, "record");
@@ -325 +362 @@
 
         } catch (IOException e) {
-            throw new RecordException("Exception occurred while updating record <" + recordId + "> in HBase table",
+            throw new RecordException("Exception occurred while updating record <" + recordId + "> on HBase table",
                     e);
         } finally {
@@ -344 +381 @@
         if (record.getRecordTypeId() == null) {
             throw new InvalidRecordException(record, "The recordType cannot be null for a record to be updated.");
-        }
-        Get get = new Get(record.getId().toBytes());
-        try {
-            if (!recordTable.exists(get)) {
-                throw new RecordNotFoundException(record);
-            }
-        } catch (IOException e) {
-            throw new RecordException("Exception occurred while retrieving original record <" + record.getId()
-                    + "> from HBase table", e);
         }
     }
@@ -703 +731 @@
             RecordException, VersionNotFoundException, TypeException {
         ArgumentValidator.notNull(recordId, "recordId");
-        Record record = newRecord();
-        record.setId(recordId);
-
-        Get get = new Get(recordId.toBytes());
-        if (requestedVersion != null) {
-            get.setMaxVersions();
-        }
-        // Add the columns for the fields to get
-        addFieldsToGet(get, fields);
-        Result result;
-        try {
-            if (!recordTable.exists(new Get(recordId.toBytes()))) {
-                throw new RecordNotFoundException(record);
-            }
-            // Retrieve the data from the repository
-            result = recordTable.get(get);
-        } catch (IOException e) {
-            throw new RecordException("Exception occurred while retrieving record <" + recordId
-                    + "> from HBase table", e);
-        }
+
+        Result result = getRow(recordId, requestedVersion, fields);
+
+        Record record = newRecord(recordId);
 
         // Set retrieved version on the record
+        extractVersion(requestedVersion, result, record);
+
+        // Extract the actual fields from the retrieved data
+        if (extractFields(result, record.getVersion(), record, readContext)) {
+            // Set the recordType explicitly in case only versioned fields were
+            // extracted
+            Pair<String, Long> recordTypePair = extractRecordType(Scope.NON_VERSIONED, result, null, record);
+            record.setRecordType(recordTypePair.getV1(), recordTypePair.getV2());
+        }
+        return record;
+    }
+
+    private void extractVersion(Long requestedVersion, Result result, Record record) throws VersionNotFoundException {
         byte[] latestVersionBytes = result.getValue(systemColumnFamilies.get(Scope.NON_VERSIONED),
                 CURRENT_VERSION_COLUMN_NAME);
@@ -736 +761 @@
             record.setVersion(latestVersion);
         }
-
-        // Extract the actual fields from the retrieved data
-        if (extractFields(result, record.getVersion(), record, readContext)) {
-            // Set the recordType explicitly in case only versioned fields were
-            // extracted
-            Pair<String, Long> recordTypePair = extractRecordType(Scope.NON_VERSIONED, result, null, record);
-            record.setRecordType(recordTypePair.getV1(), recordTypePair.getV2());
-        }
-        return record;
+    }
+
+    // Retrieves the row from the table and check if it exists and has not been flagged as deleted
+    private Result getRow(RecordId recordId, Long requestedVersion, List<FieldType> fields)
+            throws RecordNotFoundException, RecordException {
+        Result result;
+        Get get = new Get(recordId.toBytes());
+        if (requestedVersion != null) {
+            get.setMaxVersions();
+        }
+        // Add the columns for the fields to get
+        addFieldsToGet(get, fields);
+        try {
+            if (!recordTable.exists(new Get(recordId.toBytes()))) {
+                throw new RecordNotFoundException(newRecord(recordId));
+            }
+            
+            // Retrieve the data from the repository
+            result = recordTable.get(get);
+            
+            // Check if the record was deleted
+            byte[] deleted = result.getValue(systemColumnFamilies.get(Scope.NON_VERSIONED), DELETED_COLUMN_NAME);
+            if ((deleted != null) && (Bytes.toBoolean(deleted))) {
+                throw new RecordNotFoundException(newRecord(recordId));
+            }
+        } catch (IOException e) {
+            throw new RecordException("Exception occurred while retrieving record <" + recordId
+                    + "> from HBase table", e);
+        }
+        return result;
+    }
+    
+    private boolean recordExists(byte[] rowId, RowLock rowLock) throws IOException {
+        Get get = new Get(rowId, rowLock);
+        if (!recordTable.exists(get)) return false;
+        
+        get.addColumn(systemColumnFamilies.get(Scope.NON_VERSIONED), DELETED_COLUMN_NAME);
+        Result result = recordTable.get(get);
+        if (result == null || result.isEmpty()) {
+            return true;
+        } else {
+            byte[] deleted = result.getValue(systemColumnFamilies.get(Scope.NON_VERSIONED), DELETED_COLUMN_NAME);
+            if ((deleted != null) && (Bytes.toBoolean(deleted))) {
+                return false;
+            } else {
+                return true;
+            }
+        }
     }
 
@@ -829 +893 @@
     }
 
-    private void addFieldsToGet(Get get, List<FieldType> fields) throws RecordNotFoundException,
-            FieldTypeNotFoundException, RecordTypeNotFoundException, RecordException {
+    private void addFieldsToGet(Get get, List<FieldType> fields) {
         boolean added = false;
         if (fields != null) {
@@ -846 +909 @@
 
     private void addSystemColumnsToGet(Get get) {
+        get.addColumn(systemColumnFamilies.get(Scope.NON_VERSIONED), DELETED_COLUMN_NAME);
         get.addColumn(systemColumnFamilies.get(Scope.NON_VERSIONED), CURRENT_VERSION_COLUMN_NAME);
         get.addColumn(systemColumnFamilies.get(Scope.NON_VERSIONED), NON_VERSIONED_RECORDTYPEID_COLUMN_NAME);
@@ -889 +953 @@
     public void delete(RecordId recordId) throws RecordException {
         ArgumentValidator.notNull(recordId, "recordId");
-        Delete delete = new Delete(recordId.toBytes());
+        
+        org.lilycms.repository.impl.lock.RowLock rowLock = null;
+        byte[] rowId = recordId.toBytes();
         try {
-            recordTable.delete(delete);
+            // Take Custom Lock
+            rowLock = rowLocker.lockRow(rowId);
+            if (rowLock == null)
+                throw new RecordException("Failed to lock row while updating record <" + recordId
+                        + "> in HBase table", null);
+
+            if (recordExists(rowId, null)) { // Check if the record exists in the first place 
+                Put put = new Put(rowId);
+                put.add(systemColumnFamilies.get(Scope.NON_VERSIONED), DELETED_COLUMN_NAME, Bytes.toBytes(true));
+                RecordEvent recordEvent = new RecordEvent();
+                recordEvent.setType(Type.DELETE);
+                
+                RowLogMessage walMessage = wal.putMessage(recordId.toBytes(), null, recordEvent.toJsonBytes(), put);
+                if (!rowLocker.put(put, rowLock)) {
+                    throw new RecordException("Exception occurred while deleting record <" + recordId + "> on HBase table", null);
+                }
+                clearData(rowId, null);
+    
+                if (walMessage != null) {
+                    try {
+                        wal.processMessage(walMessage);
+                    } catch (RowLogException e) {
+                        // Processing the message failed, it will be retried later.
+                    }
+                }
+            }
+        } catch (RowLogException e) {
+            throw new RecordException("Exception occurred while deleting record <" + recordId
+                    + "> on HBase table", e);
+
         } catch (IOException e) {
-            throw new RecordException(
-                    "Exception occurred while deleting record <" + recordId + "> from HBase table", e);
-        }
-
-        // TODO bruno: I put this in here as a temp fix so that the Indexer would work
-        RecordEvent recordEvent = new RecordEvent();
-        recordEvent.setType(Type.DELETE);
-        try {
-            RowLogMessage walMessage = wal.putMessage(recordId.toBytes(), null, recordEvent.toJsonBytes(), null);
-            wal.processMessage(walMessage);
-            recordTable.delete(delete);
-        } catch (RowLogException e) {
-            throw new RuntimeException(e);
-        } catch (IOException e) {
-            throw new RuntimeException("Temporarily introduced second delete of record row failed.", e);
+            throw new RecordException("Exception occurred while deleting record <" + recordId + "> on HBase table",
+                    e);
+        } finally {
+            if (rowLock != null) {
+                try {
+                    rowLocker.unlockRow(rowLock);
+                } catch (IOException e) {
+                    // Ignore for now
+                }
+            }
         }
     }

projects/lily/trunk/repository/impl/src/test/java/org/lilycms/repository/impl/test/AbstractRepositoryTest.java

@@ -2 +2 @@
 
 import static org.easymock.EasyMock.createControl;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
+import static org.junit.Assert.*;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -19 +14 @@
 import org.lilycms.repository.api.*;
 import org.lilycms.repository.impl.IdGeneratorImpl;
+import org.lilycms.testfw.HBaseProxy;
 
 public abstract class AbstractRepositoryTest {
 
+    protected static final HBaseProxy HBASE_PROXY = new HBaseProxy();
     protected static IdGenerator idGenerator = new IdGeneratorImpl();
     protected static TypeManager typeManager;
@@ -636 +633 @@
         } catch (RecordNotFoundException expected) {
         }
+        try {
+            repository.update(record);
+            fail();
+        } catch (RecordNotFoundException expected) {
+        }
+        repository.delete(record.getId()); // Deleting a record twice does not throw an exception
+    }
+    
+    @Test
+    public void TestDeleteRecordCleansUpDataBeforeRecreate() throws Exception {
+        Record record = createDefaultRecord();
+        RecordId recordId = record.getId();
+        repository.delete(recordId);
+        
+     // Work around HBASE-2256
+        HBASE_PROXY.majorCompact("recordTable", "VCF");
+
+        record = repository.newRecord(recordId);
+        record.setRecordType(recordType2.getId(), recordType2.getVersion());
+        record.setField(fieldType4.getName(), 555);
+        record.setField(fieldType5.getName(), false);
+        record.setField(fieldType6.getName(), "zzz");
+        repository.create(record);
+        Record readRecord = repository.read(recordId);
+        assertEquals(Long.valueOf(1), readRecord.getVersion());
+        try {
+            readRecord.getField(fieldType1.getName());
+            fail();
+        } catch (FieldNotFoundException expected) {
+        }
+        try {
+            readRecord.getField(fieldType2.getName());
+            fail();
+        } catch (FieldNotFoundException expected) {
+        }
+        try {
+            readRecord.getField(fieldType3.getName());
+            fail();
+        } catch (FieldNotFoundException expected) {
+        }
+        
+        
+        
+        
+        
+        assertEquals(555, readRecord.getField(fieldType4.getName()));
+        assertFalse((Boolean)readRecord.getField(fieldType5.getName()));
+        assertEquals("zzz", readRecord.getField(fieldType6.getName()));
     }
 

projects/lily/trunk/repository/impl/src/test/java/org/lilycms/repository/impl/test/AvroRepositoryTest.java

@@ -36 +36 @@
 import org.lilycms.repository.impl.SizeBasedBlobStoreAccessFactory;
 import org.lilycms.repository.impl.TypeManagerRemoteImpl;
-import org.lilycms.testfw.HBaseProxy;
 import org.lilycms.testfw.TestHelper;
 
 public class AvroRepositoryTest extends AbstractRepositoryTest {
     private static HBaseRepository serverRepository;
-
-    private final static HBaseProxy HBASE_PROXY = new HBaseProxy();
 
     @BeforeClass

projects/lily/trunk/repository/impl/src/test/java/org/lilycms/repository/impl/test/HBaseRepositoryTest.java

@@ -28 +28 @@
 import org.lilycms.repository.impl.HBaseTypeManager;
 import org.lilycms.repository.impl.SizeBasedBlobStoreAccessFactory;
-import org.lilycms.testfw.HBaseProxy;
 import org.lilycms.testfw.TestHelper;
 
 public class HBaseRepositoryTest extends AbstractRepositoryTest {
 
-    private final static HBaseProxy HBASE_PROXY = new HBaseProxy();
     @BeforeClass
     public static void setUpBeforeClass() throws Exception {

projects/lily/trunk/testfw/src/main/java/org/lilycms/testfw/HBaseProxy.java

@@ -224 +224 @@
             HTable htable = new HTable(CONF, tableName);
 
-            byte[] tmpRowKey = Bytes.toBytes("HBaseProxyDummyRow");
             byte[] CF = EXPLOIT_TIMESTAMP_TABLES.get(tableName);
-            byte[] COL = Bytes.toBytes("DummyColumn");
-
-            byte[] value = null;
-            while (value == null) {
-                Put put = new Put(tmpRowKey);
-                put.add(CF, COL, 1, new byte[] { 0 });
-                htable.put(put);
-
-                Get get = new Get(tmpRowKey);
-                Result result = htable.get(get);
-                value = result.getValue(CF, COL);
-                if (value == null) {
-                    // If the value is null, it is because the delete marker has not yet been flushed/compacted away
-                    System.out.println("Waiting for flush/compact of " + tableName + " to complete");
-                }
-                Thread.sleep(100);
-            }
+            byte[] tmpRowKey = waitForCompact(tableName, CF);
 
             // Delete our dummy row again
@@ -249 +232 @@
     }
 
+    private byte[] waitForCompact(String tableName, byte[] CF) throws IOException, InterruptedException {
+        byte[] tmpRowKey = Bytes.toBytes("HBaseProxyDummyRow");
+        byte[] COL = Bytes.toBytes("DummyColumn");
+        HTable htable = new HTable(CONF, tableName);
+
+        byte[] value = null;
+        while (value == null) {
+            Put put = new Put(tmpRowKey);
+            put.add(CF, COL, 1, new byte[] { 0 });
+            htable.put(put);
+
+            Get get = new Get(tmpRowKey);
+            Result result = htable.get(get);
+            value = result.getValue(CF, COL);
+            if (value == null) {
+                // If the value is null, it is because the delete marker has not yet been flushed/compacted away
+                System.out.println("Waiting for flush/compact of " + tableName + " to complete");
+                Thread.sleep(100);
+            }
+        }
+        return tmpRowKey;
+    }
+
+    /** Force a major compaction and wait for it to finish.
+     *  This method can be used in a test to avoid issue HBASE-2256 after performing a delete operation 
+     *  Uses same principle as {@link #cleanTables()}
+     */ 
+    public void majorCompact(String tableName, String CFName) throws Exception {
+        byte[] tmpRowKey = Bytes.toBytes("HBaseProxyDummyRow");
+        byte[] CF = Bytes.toBytes(CFName);
+        byte[] COL = Bytes.toBytes("DummyColumn");
+        HBaseAdmin admin = new HBaseAdmin(getConf());
+        HTable htable = new HTable(CONF, tableName);
+        
+        // Write a dummy row
+        Put put = new Put(tmpRowKey);
+        put.add(CF, COL, 1, new byte[] { 0 });
+        htable.put(put);
+        // Delete the value again
+        Delete delete = new Delete(tmpRowKey);
+        delete.deleteColumn(CF, COL);
+        htable.delete(delete);
+        
+        // Perform major compaction
+        admin.flush(tableName);
+        admin.majorCompact(tableName);
+        
+        // Wait for compact to finish
+        waitForCompact(tableName, CF);
+    }
 }