Changeset 4654

Timestamp:

2011-02-08 11:29:57 (2 years ago)

Author:

evert

Message:

Unchanged versioned system fields are no longer stored for each version. (#162)
In case of mutable updates, old values are copied to the next version if needed (cfr sparseness of the table).

Location:

trunk/cr/repository/impl/src

Files:

• main/java/org/lilyproject/repository/impl/HBaseRepository.java (modified) (17 diffs)
• test/java/org/lilyproject/repository/impl/test/AbstractRepositoryTest.java (modified) (4 diffs)

trunk/cr/repository/impl/src/main/java/org/lilyproject/repository/impl/HBaseRepository.java

@@ -195 +195 @@
                 Result result = recordTable.get(get);
                 if (!result.isEmpty()) {
-                    byte[] oldVersion = result.getValue(systemColumnFamily, RecordColumn.VERSION.bytes);
-                    if (oldVersion != null) {
-                        version = Bytes.toLong(oldVersion) + 1;
-                        // Make sure any old data gets cleared and old blobs are deleted
-                        // This is to cover the failure scenario where a record was deleted, but a failure
-                        // occured before executing the clearData
-                        // If this was already done, this is a no-op
-                        clearData(recordId);
-                    }
+                    byte[] oldVersion = result.getValue(systemColumnFamily, RecordColumn.VERSION.bytes);
+                    if (oldVersion != null) {
+                        version = Bytes.toLong(oldVersion) + 1;
+                        // Make sure any old data gets cleared and old blobs are deleted
+                        // This is to cover the failure scenario where a record was deleted, but a failure
+                        // occured before executing the clearData
+                        // If this was already done, this is a no-op
+                        clearData(recordId);
+                    }
                 }
                 
@@ -286 +286 @@
             }
             
+            // Check if the update is an update of mutable fields
             if (updateVersion) {
                 try {
@@ -359 +360 @@
     }
 
+    // This method takes a put object containing the row's data to be updated
+    // A wal message is added to this put object
+    // The rowLocker is asked to put the data and message on the record table using the given rowlock
+    // Finally the wal is asked to process the message
     private void putRowWithWalProcessing(RecordId recordId, org.lilyproject.repository.impl.lock.RowLock rowLock,
             Put put, RecordEvent recordEvent) throws InterruptedException, RowLogException, IOException, RecordException {
@@ -464 +469 @@
     }
 
+    // Calculates which fields have changed and updates the record types of the scopes that have changed fields
     private Set<Scope> calculateChangedFields(Record record, Record originalRecord, RecordType recordType,
             Long version, Put put, RecordEvent recordEvent, Set<BlobReference> referencedBlobs, Set<BlobReference> unReferencedBlobs) throws InterruptedException, FieldTypeNotFoundException, TypeException, BlobException, RecordTypeNotFoundException, RecordException {
@@ -480 +486 @@
         // Update record types
         for (Scope scope : changedScopes) {
+            long versionOfRTField = version;
             if (Scope.NON_VERSIONED.equals(scope)) {
-                put.add(systemColumnFamily, recordTypeIdColumnNames.get(scope), 1L, Bytes.toBytes(recordType
-                        .getId()));
-                put.add(systemColumnFamily, recordTypeVersionColumnNames.get(scope), 1L, Bytes
-                        .toBytes(recordType.getVersion()));
+                versionOfRTField = 1L; // For non-versioned fields the record type is always stored at version 1.
+            }
+            // Only update the recordTypeNames and versions if they have indeed changed
+            QName originalScopeRecordTypeName = originalRecord.getRecordTypeName(scope);
+            if (originalScopeRecordTypeName == null) {
+                put.add(systemColumnFamily, recordTypeIdColumnNames.get(scope), versionOfRTField, Bytes.toBytes(recordType.getId()));
+                put.add(systemColumnFamily, recordTypeVersionColumnNames.get(scope), versionOfRTField, Bytes.toBytes(recordType.getVersion()));
             } else {
-                put.add(systemColumnFamily, recordTypeIdColumnNames.get(scope), version, Bytes
-                        .toBytes(recordType.getId()));
-                put.add(systemColumnFamily, recordTypeVersionColumnNames.get(scope), version, Bytes
-                        .toBytes(recordType.getVersion()));
-
+                RecordType originalScopeRecordType = typeManager.getRecordTypeByName(originalScopeRecordTypeName, originalRecord.getRecordTypeVersion(scope));
+                if (!recordType.getId().equals(originalScopeRecordType.getId())) {
+                    put.add(systemColumnFamily, recordTypeIdColumnNames.get(scope), versionOfRTField, Bytes.toBytes(recordType.getId()));
+                }
+                if (!recordType.getVersion().equals(originalScopeRecordType.getVersion())) {
+                    put.add(systemColumnFamily, recordTypeVersionColumnNames.get(scope), versionOfRTField, Bytes.toBytes(recordType.getVersion()));
+                }
             }
             record.setRecordType(scope, recordType.getName(), recordType.getVersion());
@@ -497 +509 @@
     }
 
+    // Returns a map (fieldname -> field) of all fields that are indicated by the record to be updated.
+    // The map also includes the fields that need to be deleted. Their name is mapped onto the delete marker.
     private Map<QName, Object> getFieldsToUpdate(Record record) {
         // Work with a copy of the map
@@ -507 +521 @@
     }
 
+    // Checks for each field if it is different from its previous value and indeed needs to be updated.
     private Set<Scope> calculateUpdateFields(Map<QName, Object> fields, Map<QName, Object> originalFields, Map<QName, Object> originalNextFields,
             Long version, Put put, RecordEvent recordEvent, Set<BlobReference> referencedBlobs, Set<BlobReference> unReferencedBlobs, boolean mutableUpdate) throws InterruptedException, FieldTypeNotFoundException, TypeException, BlobException, RecordTypeNotFoundException, RecordException {
@@ -544 +559 @@
                 } else {
                     put.add(dataColumnFamily, fieldIdAsBytes, version, encodedFieldValue);
+                    // If it is a mutable update and the next version of the field was the same as the one that is being updated,
+                    // the original value needs to be copied to that next version (due to sparseness of the table). 
                     if (originalNextFields != null && !fieldIsNewOrDeleted && originalNextFields.containsKey(fieldName)) {
                         copyValueToNextVersionIfNeeded(version, put, originalNextFields, fieldName, originalValue);
@@ -591 +608 @@
             rowLock = lockRow(recordId);
             
-            Record originalRecord = read(recordId, version, null, new ReadContext());
-
-            // Update the mutable fields
-            Put put = new Put(recordId.toBytes());
             Map<QName, Object> fields = getFieldsToUpdate(record);
             fields = filterMutableFields(fields);
+
+            Record originalRecord = read(recordId, version, null, new ReadContext());
             Map<QName, Object> originalFields = filterMutableFields(originalRecord.getFields());
+
+            Record originalNextRecord = null;
+            Map<QName, Object> originalNextFields = null;
+            try {
+                originalNextRecord = read(recordId, version + 1, null, new ReadContext());
+                originalNextFields = filterMutableFields(originalNextRecord.getFields());
+            } catch (VersionNotFoundException e) {
+                // There is no next version of the record
+            }
+
+            Put put = new Put(recordId.toBytes());
             Set<BlobReference> referencedBlobs = new HashSet<BlobReference>();
             Set<BlobReference> unReferencedBlobs = new HashSet<BlobReference>();
+
             RecordEvent recordEvent = new RecordEvent();
             recordEvent.setType(Type.UPDATE);
             recordEvent.setVersionUpdated(version);
 
-            Set<Scope> changedScopes = calculateUpdateFields(fields, originalFields, getOriginalNextFields(recordId, version), version, put, recordEvent, referencedBlobs, unReferencedBlobs, true);
+            
+            Set<Scope> changedScopes = calculateUpdateFields(fields, originalFields, originalNextFields, version, put, recordEvent, referencedBlobs, unReferencedBlobs, true);
             for (BlobReference referencedBlob : referencedBlobs) {
                 referencedBlob.setRecordId(recordId);
@@ -613 +641 @@
             
             if (!changedScopes.isEmpty()) {
+                // Update the record types
+                
                 // If no record type is specified explicitly, use the current one of the non-versioned scope
                 QName recordTypeName = record.getRecordTypeName() != null ? record.getRecordTypeName() : originalRecord.getRecordTypeName();
@@ -625 +655 @@
                 RecordType recordType = typeManager.getRecordTypeByName(recordTypeName, recordTypeVersion);
                 
-                // Update the mutable record type
-                Scope scope = Scope.VERSIONED_MUTABLE;
-                newRecord.setRecordType(scope, recordType.getName(), recordType.getVersion());
+                // Update the mutable record type in the record object
+                Scope mutableScope = Scope.VERSIONED_MUTABLE;
+                newRecord.setRecordType(mutableScope, recordType.getName(), recordType.getVersion());
                 
+                // If the record type changed, update it on the record table
+                QName originalMutableScopeRecordTypeName = originalRecord.getRecordTypeName(mutableScope);
+                if (originalMutableScopeRecordTypeName == null) { // There was no initial mutable record type yet
+                    put.add(systemColumnFamily, recordTypeIdColumnNames.get(mutableScope), version, Bytes.toBytes(recordType.getId()));
+                    put.add(systemColumnFamily, recordTypeVersionColumnNames.get(mutableScope), version, Bytes.toBytes(recordType.getVersion()));
+                } else {
+                    RecordType originalMutableScopeRecordType = typeManager.getRecordTypeByName(originalMutableScopeRecordTypeName, originalRecord.getRecordTypeVersion(mutableScope));
+                    if (!recordType.getId().equals(originalMutableScopeRecordType.getId())) {
+                        // If the next record version had the same record type name, copy the original value to that one
+                        if (originalNextRecord != null && originalMutableScopeRecordType.getName().equals(originalNextRecord.getRecordTypeName(mutableScope))) {
+                            put.add(systemColumnFamily, recordTypeIdColumnNames.get(mutableScope), version+1, Bytes.toBytes(originalMutableScopeRecordType.getId()));
+                        }
+                        put.add(systemColumnFamily, recordTypeIdColumnNames.get(mutableScope), version, Bytes.toBytes(recordType.getId()));
+                    }
+                    if (!recordType.getVersion().equals(originalMutableScopeRecordType.getVersion())) {
+                        // If the next record version had the same record type version, copy the original value to that one
+                        if (originalNextRecord != null && originalMutableScopeRecordType.getVersion().equals(originalNextRecord.getRecordTypeVersion(mutableScope))) {
+                            put.add(systemColumnFamily, recordTypeIdColumnNames.get(mutableScope), version+1, Bytes.toBytes(originalMutableScopeRecordType.getVersion()));
+                        }
+                        put.add(systemColumnFamily, recordTypeVersionColumnNames.get(mutableScope), version, Bytes.toBytes(recordType.getVersion()));
+                    }
+                }
+                
+                // Validate if the new values for the record are valid wrt the recordType (e.g. mandatory fields)
                 validateRecord(newRecord, originalRecord, recordType);
-
-                put.add(systemColumnFamily, recordTypeIdColumnNames.get(scope), version, Bytes
-                        .toBytes(recordType.getId()));
-                put.add(systemColumnFamily, recordTypeVersionColumnNames.get(scope), version, Bytes
-                        .toBytes(recordType.getVersion()));
 
                 recordEvent.setVersionUpdated(version);
@@ -701 +750 @@
     }
 
-    private Map<QName, Object> getOriginalNextFields(RecordId recordId, Long version)
-            throws RecordTypeNotFoundException, FieldTypeNotFoundException, RecordException,
-            TypeException, RecordNotFoundException, InterruptedException {
-        try {
-            Record originalNextRecord = read(recordId, version + 1, null, new ReadContext());
-            return filterMutableFields(originalNextRecord.getFields());
-        } catch (VersionNotFoundException exception) {
-            // There is no next record
-            return null;
-        }
-    }
-
     public Record read(RecordId recordId) throws RecordNotFoundException, RecordTypeNotFoundException,
             FieldTypeNotFoundException, RecordException, VersionNotFoundException, TypeException,
@@ -844 +881 @@
             // extracted
             Pair<String, Long> recordTypePair = extractRecordType(Scope.NON_VERSIONED, result, null, record);
-            RecordType recordType = typeManager.getRecordTypeById(recordTypePair.getV1(), recordTypePair.getV2());
-            record.setRecordType(recordType.getName(), recordType.getVersion());
-            readContext.setRecordTypeId(Scope.NON_VERSIONED, recordType);
+            if (recordTypePair != null) {
+                RecordType recordType = typeManager.getRecordTypeById(recordTypePair.getV1(), recordTypePair.getV2());
+                record.setRecordType(recordType.getName(), recordType.getVersion());
+                readContext.setRecordTypeId(Scope.NON_VERSIONED, recordType);
+            }
         }
         return record;
@@ -915 +954 @@
         if (version == null) {
             // Get latest version
-            return new Pair<String, Long>(Bytes.toString(result.getValue(systemColumnFamily,
-                    recordTypeIdColumnNames.get(scope))), Bytes.toLong(result.getValue(systemColumnFamily,
-                    recordTypeVersionColumnNames.get(scope))));
+            byte[] idBytes = result.getValue(systemColumnFamily,
+                    recordTypeIdColumnNames.get(scope));
+            byte[] versionBytes = result.getValue(systemColumnFamily,
+                    recordTypeVersionColumnNames.get(scope));
+            if ((idBytes == null || idBytes.length == 0) || (versionBytes == null || versionBytes.length == 0))
+                return null;
+            return new Pair<String, Long>(Bytes.toString(idBytes), Bytes.toLong(versionBytes));
 
         } else {
@@ -932 +975 @@
         byte[] recordTypeIdColumnName = recordTypeIdColumnNames.get(scope);
         byte[] recordTypeVersionColumnName = recordTypeVersionColumnNames.get(scope);
-        Entry<Long, byte[]> ceilingEntry = versionableSystemCFversions.get(recordTypeIdColumnName)
-                .ceilingEntry(version);
+        Entry<Long, byte[]> idCeilingEntry = null;
+        NavigableMap<Long, byte[]> recordTypeIdMap = versionableSystemCFversions.get(recordTypeIdColumnName);
+        if (recordTypeIdMap != null) {
+            idCeilingEntry = recordTypeIdMap.ceilingEntry(version);
+        }
         String recordTypeId = null;
-        if (ceilingEntry != null) {
-            recordTypeId = Bytes.toString(ceilingEntry.getValue());
-        }
+        if (idCeilingEntry == null)
+            return null;
+        recordTypeId = Bytes.toString(idCeilingEntry.getValue());
+        
         Long recordTypeVersion = null;
-        ceilingEntry = versionableSystemCFversions.get(recordTypeVersionColumnName).ceilingEntry(version);
-        if (ceilingEntry != null) {
-            recordTypeVersion = Bytes.toLong(ceilingEntry.getValue());
-        }
+        Entry<Long, byte[]> versionCeilingEntry = null;
+        NavigableMap<Long, byte[]> recordTypeVersionMap = versionableSystemCFversions.get(recordTypeVersionColumnName);
+        if (recordTypeVersionMap != null) {
+            versionCeilingEntry = recordTypeVersionMap.ceilingEntry(version);
+        }
+        if (versionCeilingEntry == null)
+            return null;
+        recordTypeVersion = Bytes.toLong(versionCeilingEntry.getValue());
         Pair<String, Long> recordType = new Pair<String, Long>(recordTypeId, recordTypeVersion);
         return recordType;
@@ -1021 +1072 @@
             record.setField(field.getV1(), field.getV2());
         }
-        for (Scope readScope : context.getScopes()) {
-            Pair<String, Long> recordTypePair = extractRecordType(readScope, result, version, record);
-            RecordType recordType = typeManager.getRecordTypeById(recordTypePair.getV1(), recordTypePair.getV2());
-            record.setRecordType(readScope, recordType.getName(), recordType.getVersion());
-            context.setRecordTypeId(readScope, recordType);
+        for (Scope scope : Scope.values()) {
+            Pair<String, Long> recordTypePair = extractRecordType(scope, result, version, record);
+            if (recordTypePair != null) {
+                RecordType recordType = typeManager.getRecordTypeById(recordTypePair.getV1(), recordTypePair.getV2());
+                record.setRecordType(scope, recordType.getName(), recordType.getVersion());
+                context.setRecordTypeId(scope, recordType);
+            }
         }
         return true;
@@ -1167 +1220 @@
 
     private Set<BlobReference> getReferencedBlobs(FieldTypeImpl fieldType, Object value) throws BlobException {
-        fieldType.getValueType().getValues(value);
         HashSet<BlobReference> referencedBlobs = new HashSet<BlobReference>();
         ValueType valueType = fieldType.getValueType();

trunk/cr/repository/impl/src/test/java/org/lilyproject/repository/impl/test/AbstractRepositoryTest.java

@@ -900 +900 @@
     @Test
     public void testUpdateMutableFieldWithNewRecordType() throws Exception {
-        Record record = createDefaultRecord();
+        // Create default record
+        Record record = createDefaultRecord();
+        
+        // Update the record, creates a second version
         Record updateRecord = record.clone();
         updateRecord.setField(fieldType1.getName(), "value2");
         updateRecord.setField(fieldType2.getName(), 789);
         updateRecord.setField(fieldType3.getName(), false);
-        repository.update(updateRecord);
-
+        Record updatedRecord = repository.update(updateRecord, false, false);
+
+        // Read the first version of the record
+        Record readRecord = repository.read(record.getId(), Long.valueOf(1));
+        assertEquals(recordType1.getName(), readRecord.getRecordTypeName());
+        assertEquals(recordType1.getVersion(), readRecord.getRecordTypeVersion());
+        assertEquals(recordType1.getName(), readRecord.getRecordTypeName(Scope.NON_VERSIONED));
+        assertEquals(recordType1.getVersion(), readRecord.getRecordTypeVersion(Scope.NON_VERSIONED));
+        assertEquals(recordType1.getName(), readRecord.getRecordTypeName(Scope.VERSIONED));
+        assertEquals(recordType1.getVersion(), readRecord.getRecordTypeVersion(Scope.VERSIONED));
+        assertEquals(recordType1.getName(), readRecord.getRecordTypeName(Scope.VERSIONED_MUTABLE));
+        assertEquals(recordType1.getVersion(), readRecord.getRecordTypeVersion(Scope.VERSIONED_MUTABLE));
+        
+        // Do a mutable update of the first version of the record, change the record type
         Record updateMutableRecord = repository.newRecord(record.getId());
         updateMutableRecord.setVersion(Long.valueOf(1));
@@ -914 +929 @@
         updateMutableRecord.setField(fieldType6.getName(), "value3");
 
-        assertEquals(Long.valueOf(1), repository.update(updateMutableRecord, true, false).getVersion());
-
-        Record readRecord = repository.read(record.getId(), Long.valueOf(1));
+        Record updatedMutableRecord = repository.update(updateMutableRecord, true, false);
+        assertEquals(Long.valueOf(1), updatedMutableRecord.getVersion());
+
+        // Read the first version of the record again
+        readRecord = repository.read(record.getId(), Long.valueOf(1));
         assertEquals(Long.valueOf(1), readRecord.getVersion());
         assertEquals("value2", readRecord.getField(fieldType1.getName()));
@@ -938 +955 @@
         }
         assertEquals(recordType1.getName(), readRecord.getRecordTypeName());
+        assertEquals(recordType1.getVersion(), readRecord.getRecordTypeVersion());
         assertEquals(recordType1.getName(), readRecord.getRecordTypeName(Scope.NON_VERSIONED));
+        assertEquals(recordType1.getVersion(), readRecord.getRecordTypeVersion(Scope.NON_VERSIONED));
         assertEquals(recordType1.getName(), readRecord.getRecordTypeName(Scope.VERSIONED));
+        assertEquals(recordType1.getVersion(), readRecord.getRecordTypeVersion(Scope.VERSIONED));
+        // The mutable record type should have been changed
         assertEquals(recordType2.getName(), readRecord.getRecordTypeName(Scope.VERSIONED_MUTABLE));
-
+        assertEquals(recordType2.getVersion(), readRecord.getRecordTypeVersion(Scope.VERSIONED_MUTABLE));
+
+        // Read the second version again of the record
         readRecord = repository.read(record.getId());
         assertEquals(Long.valueOf(2), readRecord.getVersion());
@@ -951 +974 @@
         assertEquals(recordType1.getName(), readRecord.getRecordTypeName(Scope.NON_VERSIONED));
         assertEquals(recordType1.getName(), readRecord.getRecordTypeName(Scope.VERSIONED));
+        // The original mutable record type should have been copied to the next version
         assertEquals(recordType1.getName(), readRecord.getRecordTypeName(Scope.VERSIONED_MUTABLE));
     }