diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/metadata/IndexTypes.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/metadata/IndexTypes.java index 8d10f26d9e..eda844ffb4 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/metadata/IndexTypes.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/metadata/IndexTypes.java @@ -169,6 +169,11 @@ public class IndexTypes { */ public static final String VECTOR = "vector"; + /** + * An index with record change history. + */ + public static final String RECORD_CHANGES = "record_changes"; + private IndexTypes() { } } diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/RecordChangesIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/RecordChangesIndexMaintainer.java new file mode 100644 index 0000000000..72d7fd2c58 --- /dev/null +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/RecordChangesIndexMaintainer.java @@ -0,0 +1,284 @@ +/* + * RecordChangesIndexMaintainer.java + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2026 Apple Inc. and the FoundationDB project authors + * + * Licensed 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 com.apple.foundationdb.record.provider.foundationdb.indexes; + +import com.apple.foundationdb.KeyValue; +import com.apple.foundationdb.annotation.API; +import com.apple.foundationdb.async.AsyncUtil; +import com.apple.foundationdb.record.EndpointType; +import com.apple.foundationdb.record.IndexEntry; +import com.apple.foundationdb.record.IndexScanType; +import com.apple.foundationdb.record.RecordChangesProto; +import com.apple.foundationdb.record.RecordCoreException; +import com.apple.foundationdb.record.RecordCursor; +import com.apple.foundationdb.record.RecordMetaData; +import com.apple.foundationdb.record.ScanProperties; +import com.apple.foundationdb.record.TupleRange; +import com.apple.foundationdb.record.metadata.RecordType; +import com.apple.foundationdb.record.metadata.expressions.KeyExpression; +import com.apple.foundationdb.record.metadata.expressions.VersionKeyExpression; +import com.apple.foundationdb.record.provider.foundationdb.FDBIndexableRecord; +import com.apple.foundationdb.record.provider.foundationdb.FDBIndexedRawRecord; +import com.apple.foundationdb.record.provider.foundationdb.FDBRawRecord; +import com.apple.foundationdb.record.provider.foundationdb.FDBRecordVersion; +import com.apple.foundationdb.record.provider.foundationdb.FDBStoreTimer; +import com.apple.foundationdb.record.provider.foundationdb.FDBStoredRecord; +import com.apple.foundationdb.record.provider.foundationdb.FDBStoredRecordBuilder; +import com.apple.foundationdb.record.provider.foundationdb.IndexMaintainerState; +import com.apple.foundationdb.record.provider.foundationdb.IndexScanBounds; +import com.apple.foundationdb.record.provider.foundationdb.IndexScrubbingTools; +import com.apple.foundationdb.record.provider.foundationdb.KeyValueCursor; +import com.apple.foundationdb.record.provider.foundationdb.SplitHelper; +import com.apple.foundationdb.tuple.Tuple; +import com.google.protobuf.Descriptors; +import com.google.protobuf.InvalidProtocolBufferException; +import com.google.protobuf.Message; +import com.google.protobuf.ZeroCopyByteString; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.util.BitSet; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +/** + * An index that records changes to a record on save. + * The keys of the index would normally include {@link VersionKeyExpression Version}, + * so that the index represents a complete history. + */ +@API(API.Status.EXPERIMENTAL) +public class RecordChangesIndexMaintainer extends StandardIndexMaintainer { + public record RecordChangesEntry(@Nonnull Tuple key, + @Nullable FDBStoredRecord oldRecord, + @Nullable FDBStoredRecord newRecord) { + } + + private final int versionPosition; + + protected RecordChangesIndexMaintainer(IndexMaintainerState state) { + super(state); + final List keys = state.index.getRootExpression().normalizeKeyForPositions(); + int found = -1; + for (int i = 0; i < keys.size(); i++) { + if (keys.get(i) instanceof VersionKeyExpression) { + found = i; + break; + } + } + versionPosition = found; + } + + @Nonnull + @SuppressWarnings("PMD.CloseResource") + public RecordCursor> scanChanges(@Nonnull TupleRange range, + @Nullable byte[] continuation, + @Nonnull ScanProperties scanProperties) { + final RecordCursor keyValues = KeyValueCursor.Builder.newBuilder(state.indexSubspace) + .setContext(state.context) + .setRange(range) + .setContinuation(continuation) + .setScanProperties(scanProperties) + .build(); + return new SplitHelper.KeyValueUnsplitter(state.context, state.indexSubspace, keyValues, false, null, scanProperties) + .map(this::deserializeChange); + } + + @Nonnull + @SuppressWarnings("PMD.CloseResource") + public RecordCursor> scanChanges(@Nonnull TupleRange range, + @Nullable Tuple lastKey, + @Nonnull ScanProperties scanProperties) { + final RecordCursor keyValues = KeyValueCursor.Builder.newBuilder(state.indexSubspace) + .setContext(state.context) + .setLow(lastKey == null ? range.getLow() : lastKey, + lastKey == null ? range.getLowEndpoint() : EndpointType.RANGE_EXCLUSIVE) + .setHigh(range.getHigh(), range.getHighEndpoint()) + .setScanProperties(scanProperties) + .build(); + return new SplitHelper.KeyValueUnsplitter(state.context, state.indexSubspace, keyValues, false, null, scanProperties) + .map(this::deserializeChange); + } + + @Override + @Nonnull + public CompletableFuture update(@Nullable final FDBIndexableRecord oldRecord, + @Nullable final FDBIndexableRecord newRecord) { + // No entries are removed. Key should normally be the same, up to version, for both records. + final FDBIndexableRecord indexRecord; + if (newRecord != null) { + if (newRecord.hasVersion()) { + indexRecord = newRecord; + } else { + indexRecord = ((FDBStoredRecord)newRecord).withVersion(FDBRecordVersion.incomplete(state.context.claimLocalVersion())); + } + } else if (oldRecord != null) { + indexRecord = ((FDBStoredRecord)oldRecord).withVersion(FDBRecordVersion.incomplete(state.context.claimLocalVersion())); + } else { + return AsyncUtil.DONE; + } + final List indexEntries = filteredIndexEntries(indexRecord); + if (indexEntries == null || indexEntries.isEmpty()) { + return AsyncUtil.DONE; + } + final byte[] serializedChange = serializeChange(state.store.getRecordMetaData(), indexRecord.getRecordType(), indexRecord.getPrimaryKey(), + newRecord == null ? null : newRecord.getRecord(), + oldRecord == null ? null : oldRecord.getRecord(), + oldRecord == null ? null : oldRecord.getVersion()); + for (IndexEntry indexEntry : indexEntries) { + if (!indexEntry.getValue().isEmpty()) { + throw new RecordCoreException("Index entry has conflicting value."); + } + final long startTime = System.nanoTime(); + final Tuple entryKey = indexEntryKey(indexEntry.getKey(), indexRecord.getPrimaryKey()); + // It might be simpler to pass indexRecord.getVersion(), but version cannot be in both key and value. + SplitHelper.saveWithSplit(state.context, state.indexSubspace, entryKey, serializedChange, null); + if (state.store.getTimer() != null) { + state.store.getTimer().recordSinceNanoTime(FDBStoreTimer.Events.SAVE_INDEX_ENTRY, startTime); + } + } + return AsyncUtil.DONE; + } + + @Override + @Nonnull + public RecordCursor scan(@Nonnull IndexScanType scanType, @Nonnull TupleRange range, + @Nullable byte[] continuation, + @Nonnull ScanProperties scanProperties) { + throw new RecordCoreException("Cannot scan change index directly."); + } + + @Nonnull + @Override + public RecordCursor scanRemoteFetch(@Nonnull final IndexScanBounds scanBounds, + @Nullable final byte[] continuation, + @Nonnull final ScanProperties scanProperties, + int commonPrimaryKeyLength) { + throw new RecordCoreException("Cannot scan change index remotely."); + } + + @Nullable + @Override + public IndexScrubbingTools getIndexScrubbingTools(final IndexScrubbingTools.ScrubbingType type) { + return null; + } + + @Nonnull + private byte[] serializeChange(@Nonnull RecordMetaData metaData, @Nonnull RecordType recordType, @Nonnull Tuple primaryKey, + @Nullable Message newRecord, @Nullable Message oldRecord, @Nullable FDBRecordVersion oldVersion) { + RecordChangesProto.RecordChange.Builder change = RecordChangesProto.RecordChange.newBuilder() + .setPrimaryKey(ZeroCopyByteString.wrap(primaryKey.pack())); + if (newRecord == null) { + if (oldRecord != null) { + change.setOldRecord(ZeroCopyByteString.wrap(state.store.getSerializer().serialize(metaData, recordType, oldRecord, state.context.getTimer()))); + } + } else { + change.setNewRecord(ZeroCopyByteString.wrap(state.store.getSerializer().serialize(metaData, recordType, newRecord, state.context.getTimer()))); + if (oldRecord != null) { + Message.Builder changedBuilder = oldRecord.toBuilder(); + for (Map.Entry fieldEntry : newRecord.getAllFields().entrySet()) { + Descriptors.FieldDescriptor fieldDescriptor = fieldEntry.getKey(); + if (!changedBuilder.hasField(fieldDescriptor)) { + change.addChangedEmptyFields(fieldDescriptor.getIndex()); + } else if (changedBuilder.getField(fieldDescriptor).equals(fieldEntry.getValue())) { + changedBuilder.clearField(fieldDescriptor); + } + } + change.setOldRecord(ZeroCopyByteString.wrap(state.store.getSerializer().serialize(metaData, recordType, changedBuilder.build(), state.context.getTimer()))); + } + } + if (oldVersion != null) { + change.setOldVersion(ZeroCopyByteString.wrap(oldVersion.toBytes(false))); + } + return change.build().toByteArray(); + } + + @Nonnull + private RecordChangesEntry deserializeChange(@Nonnull FDBRawRecord unsplit) { + final Tuple key = unsplit.getPrimaryKey(); + final FDBRecordVersion version; + if (versionPosition < 0) { + version = null; + } else { + version = FDBRecordVersion.fromVersionstamp(key.getVersionstamp(versionPosition)); + } + final RecordChangesProto.RecordChange changeProto; + try { + changeProto = RecordChangesProto.RecordChange.parseFrom(unsplit.getRawRecord()); + } catch (InvalidProtocolBufferException ex) { + throw new RecordCoreException("error parsing change entry", ex); + } + final RecordMetaData metaData = state.store.getRecordMetaData(); + final Tuple primaryKey = Tuple.fromBytes(changeProto.getPrimaryKey().toByteArray()); + final FDBStoredRecord newRecord; + if (changeProto.hasNewRecord()) { + final Message protoRecord = state.store.getSerializer().deserialize(metaData, key, + changeProto.getNewRecord().toByteArray(), state.context.getTimer()); + final FDBStoredRecordBuilder recordBuilder = FDBStoredRecord.newBuilder(protoRecord).setPrimaryKey(primaryKey); + final RecordType recordType = metaData.getRecordTypeForDescriptor(protoRecord.getDescriptorForType()); + recordBuilder.setRecordType(recordType); + if (metaData.isStoreRecordVersions()) { + // If we could have it as inline value, this would be unsplit.getVersion(). + recordBuilder.setVersion(version); + } + newRecord = recordBuilder.build(); + } else { + newRecord = null; + } + final FDBStoredRecord oldRecord; + if (changeProto.hasOldRecord()) { + final Message protoRecord = state.store.getSerializer().deserialize(metaData, key, + changeProto.getOldRecord().toByteArray(), state.context.getTimer()); + final FDBStoredRecordBuilder recordBuilder = FDBStoredRecord.newBuilder().setPrimaryKey(primaryKey); + final RecordType recordType = metaData.getRecordTypeForDescriptor(protoRecord.getDescriptorForType()); + recordBuilder.setRecordType(recordType); + if (newRecord == null) { + recordBuilder.setRecord(protoRecord); + } else { + Message.Builder protoBuilder = protoRecord.toBuilder(); + BitSet changedEmpty = new BitSet(); + for (Integer fieldIndex : changeProto.getChangedEmptyFieldsList()) { + changedEmpty.set(fieldIndex); + } + // Fill in unchanged fields. + for (Map.Entry fieldEntry : newRecord.getRecord().getAllFields().entrySet()) { + Descriptors.FieldDescriptor fieldDescriptor = fieldEntry.getKey(); + if (!protoBuilder.hasField(fieldDescriptor) && !changedEmpty.get(fieldDescriptor.getIndex())) { + protoBuilder.setField(fieldDescriptor, fieldEntry.getValue()); + } + } + recordBuilder.setRecord(protoBuilder.build()); + } + if (changeProto.hasOldVersion()) { + FDBRecordVersion recordVersion = FDBRecordVersion.fromBytes(changeProto.getOldVersion().toByteArray()); + if (!recordVersion.isComplete()) { + // Not able to complete it in the middle of the serialized protobuf, so deferred to here. + recordVersion = recordVersion.withCommittedVersion(version.getGlobalVersion()); + } + recordBuilder.setVersion(recordVersion); + } + oldRecord = recordBuilder.build(); + } else { + oldRecord = null; + } + return new RecordChangesEntry<>(key, oldRecord, newRecord); + } +} diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/RecordChangesIndexMaintainerFactory.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/RecordChangesIndexMaintainerFactory.java new file mode 100644 index 0000000000..8c0be38304 --- /dev/null +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/RecordChangesIndexMaintainerFactory.java @@ -0,0 +1,90 @@ +/* + * RecordChangesIndexMaintainerFactory.java + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2026 Apple Inc. and the FoundationDB project authors + * + * Licensed 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 com.apple.foundationdb.record.provider.foundationdb.indexes; + +import com.apple.foundationdb.annotation.API; +import com.apple.foundationdb.record.RecordMetaData; +import com.apple.foundationdb.record.metadata.Index; +import com.apple.foundationdb.record.metadata.IndexTypes; +import com.apple.foundationdb.record.metadata.IndexValidator; +import com.apple.foundationdb.record.metadata.MetaDataValidator; +import com.apple.foundationdb.record.provider.foundationdb.IndexGeneralAttributes; +import com.apple.foundationdb.record.provider.foundationdb.IndexMaintainer; +import com.apple.foundationdb.record.provider.foundationdb.IndexMaintainerFactory; +import com.apple.foundationdb.record.provider.foundationdb.IndexMaintainerState; +import com.apple.foundationdb.record.query.plan.cascades.MatchCandidate; +import com.google.auto.service.AutoService; + +import javax.annotation.Nonnull; +import java.util.Arrays; +import java.util.List; + +/** + * A factory for {@link RecordChangesIndexMaintainer} indexes. + */ +@AutoService(IndexMaintainerFactory.class) +@API(API.Status.EXPERIMENTAL) +public class RecordChangesIndexMaintainerFactory implements IndexMaintainerFactory { + static final String[] TYPES = { IndexTypes.RECORD_CHANGES }; + private static final IndexGeneralAttributes GENERAL_ATTRIBUTES = new IndexGeneralAttributes(true); + + @Override + @Nonnull + public Iterable getIndexTypes() { + return Arrays.asList(TYPES); + } + + @Nonnull + @Override + public IndexValidator getIndexValidator(Index forIndex) { + return new IndexValidator(forIndex) { + @Override + public void validate(@Nonnull MetaDataValidator metaDataValidator) { + super.validate(metaDataValidator); + validateNotGrouping(); + validateNotUnique(); + // Does not _require_ Version(), but makes more sense with it. + if (index.getRootExpression().versionColumns() > 0) { + validateStoresRecordVersions(metaDataValidator); + validateVersionKey(); + } + } + }; + } + + @Override + @Nonnull + public IndexMaintainer getIndexMaintainer(@Nonnull IndexMaintainerState state) { + return new RecordChangesIndexMaintainer(state); + } + + @Nonnull + @Override + public Iterable createMatchCandidates(@Nonnull final RecordMetaData metaData, @Nonnull final Index index, final boolean reverse) { + return List.of(); + } + + @Nonnull + @Override + public IndexGeneralAttributes getIndexGeneralAttributes(@Nonnull final Index index) { + return GENERAL_ATTRIBUTES; + } +} diff --git a/fdb-record-layer-core/src/main/proto/record_changes.proto b/fdb-record-layer-core/src/main/proto/record_changes.proto new file mode 100644 index 0000000000..045a2d1aeb --- /dev/null +++ b/fdb-record-layer-core/src/main/proto/record_changes.proto @@ -0,0 +1,33 @@ +/* + * record_changes.proto + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2026 Apple Inc. and the FoundationDB project authors + * + * Licensed 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. + */ +syntax = "proto2"; + +package com.apple.foundationdb.record; +import "google/protobuf/descriptor.proto"; + +option java_outer_classname = "RecordChangesProto"; + +message RecordChange { + optional bytes primary_key = 1; + optional bytes new_record = 2; + optional bytes old_record = 3; + optional bytes old_version = 4; + repeated int32 changed_empty_fields = 5; // As opposed to unchanged. +} diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/RecordChangesIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/RecordChangesIndexTest.java new file mode 100644 index 0000000000..8445f1ed0c --- /dev/null +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/RecordChangesIndexTest.java @@ -0,0 +1,192 @@ +/* + * RecordChangesIndexTest.java + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2026 Apple Inc. and the FoundationDB project authors + * + * Licensed 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 com.apple.foundationdb.record.provider.foundationdb.indexes; + +import com.apple.foundationdb.record.ScanProperties; +import com.apple.foundationdb.record.TestRecords1Proto.MySimpleRecord; +import com.apple.foundationdb.record.TupleRange; +import com.apple.foundationdb.record.metadata.Index; +import com.apple.foundationdb.record.metadata.IndexTypes; +import com.apple.foundationdb.record.metadata.expressions.VersionKeyExpression; +import com.apple.foundationdb.record.provider.foundationdb.FDBRecordContext; +import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreTestBase; +import com.apple.foundationdb.record.provider.foundationdb.FDBStoredRecord; +import com.apple.foundationdb.record.provider.foundationdb.FDBStoredRecordBuilder; +import com.apple.foundationdb.record.provider.foundationdb.SplitHelper; +import com.apple.foundationdb.tuple.Tuple; +import com.apple.test.Tags; +import com.google.protobuf.Message; +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.util.List; +import java.util.function.Function; + +/** + * Tests for {@code RECORD_CHANGES} type indexes. + */ +@Tag(Tags.RequiresFDB) +class RecordChangesIndexTest extends FDBRecordStoreTestBase { + + @Nonnull + private final RecordMetaDataHook globalVersionIndex = + metaDataBuilder -> { + metaDataBuilder.setSplitLongRecords(true); + metaDataBuilder.addUniversalIndex(new Index("globalChanges", VersionKeyExpression.VERSION, IndexTypes.RECORD_CHANGES)); + }; + + @Test + void basicGlobalIndex() { + MySimpleRecord.Builder record1 = MySimpleRecord.newBuilder().setRecNo(1066L).setNumValue2(1); + MySimpleRecord.Builder record2 = MySimpleRecord.newBuilder().setRecNo(1067L).setNumValue2(2); + + FDBStoredRecord record1_1; + FDBStoredRecord record2_1; + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, globalVersionIndex); + record1_1 = recordStore.saveRecord(record1.build()); + record2_1 = recordStore.saveRecord(record2.build()); + context.commit(); + record1_1 = clearSize(record1_1.withCommittedVersion(context.getVersionStamp())); + record2_1 = clearSize(record2_1.withCommittedVersion(context.getVersionStamp())); + } + + Tuple lastKey = null; + + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, globalVersionIndex); + List> changes = getSimpleChanges(lastKey); + Assertions.assertThat(changes).hasSize(2); + Assertions.assertThat(changes.get(0).oldRecord()).isNull(); + Assertions.assertThat(changes.get(0).newRecord()).isEqualTo(record1_1); + Assertions.assertThat(changes.get(1).oldRecord()).isNull(); + Assertions.assertThat(changes.get(1).newRecord()).isEqualTo(record2_1); + lastKey = changes.get(1).key(); + } + + record1.setNumValue2(3).setStrValueIndexed("hello"); + + FDBStoredRecord record1_2; + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, globalVersionIndex); + record1_2 = recordStore.saveRecord(record1.build()); + context.commit(); + record1_2 = clearSize(record1_2.withCommittedVersion(context.getVersionStamp())); + } + + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, globalVersionIndex); + List> changes = getSimpleChanges(lastKey); + Assertions.assertThat(changes).hasSize(1); + Assertions.assertThat(changes.get(0).oldRecord()).isEqualTo(record1_1); + Assertions.assertThat(changes.get(0).newRecord()).isEqualTo(record1_2); + lastKey = changes.get(0).key(); + } + + record2.clearNumValue2(); + + FDBStoredRecord record2_2; + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, globalVersionIndex); + record2_2 = recordStore.saveRecord(record2.build()); + context.commit(); + record2_2 = clearSize(record2_2.withCommittedVersion(context.getVersionStamp())); + } + + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, globalVersionIndex); + List> changes = getSimpleChanges(lastKey); + Assertions.assertThat(changes).hasSize(1); + Assertions.assertThat(changes.get(0).oldRecord()).isEqualTo(record2_1); + Assertions.assertThat(changes.get(0).newRecord()).isEqualTo(record2_2); + lastKey = changes.get(0).key(); + } + + } + + @Test + void multipleChangesSameRecord() { + MySimpleRecord.Builder record1 = MySimpleRecord.newBuilder().setRecNo(1066L).setNumValue2(1); + + FDBStoredRecord record1_1; + FDBStoredRecord record1_2; + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, globalVersionIndex); + record1_1 = recordStore.saveRecord(record1.build()); + record1.setStrValueIndexed("hello"); + // This will make a change with an incomplete old record version. + record1_2 = recordStore.saveRecord(record1.build()); + context.commit(); + record1_1 = clearSize(record1_1.withCommittedVersion(context.getVersionStamp())); + record1_2 = clearSize(record1_2.withCommittedVersion(context.getVersionStamp())); + } + + Tuple lastKey = null; + + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, globalVersionIndex); + List> changes = getSimpleChanges(lastKey); + Assertions.assertThat(changes).hasSize(2); + Assertions.assertThat(changes.get(0).oldRecord()).isNull(); + Assertions.assertThat(changes.get(0).newRecord()).isEqualTo(record1_1); + Assertions.assertThat(changes.get(1).oldRecord()).isEqualTo(record1_1); + Assertions.assertThat(changes.get(1).newRecord()).isEqualTo(record1_2); + lastKey = changes.get(1).key(); + } + + } + + private FDBStoredRecord clearSize(@Nonnull FDBStoredRecord storedRecord) { + return storedRecord.asBuilder().setSize(new SplitHelper.SizeInfo()).build(); + } + + private List> getSimpleChanges(@Nullable Tuple lastKey) { + final Index index = recordStore.getRecordMetaData().getIndex("globalChanges"); + RecordChangesIndexMaintainer indexMaintainer = (RecordChangesIndexMaintainer)recordStore.getIndexMaintainer(index); + return indexMaintainer.scanChanges(TupleRange.ALL, lastKey, ScanProperties.FORWARD_SCAN) + .map(entry -> narrowChange(entry, r -> MySimpleRecord.newBuilder().mergeFrom(r).build())) + .asList().join(); + } + + private RecordChangesIndexMaintainer.RecordChangesEntry narrowChange(@Nonnull RecordChangesIndexMaintainer.RecordChangesEntry raw, + @Nonnull Function narrowFunction) { + return new RecordChangesIndexMaintainer.RecordChangesEntry<>(raw.key(), + narrowStoredRecord(raw.oldRecord(), narrowFunction), + narrowStoredRecord(raw.newRecord(), narrowFunction)); + } + + @SuppressWarnings("unchecked") + @Nullable + private FDBStoredRecord narrowStoredRecord(@Nullable FDBStoredRecord storedRecord, + @Nonnull Function narrowFunction) { + if (storedRecord == null) { + return null; + } + FDBStoredRecordBuilder builder = storedRecord.asBuilder(); + FDBStoredRecordBuilder narrowBuilder = (FDBStoredRecordBuilder)builder; + narrowBuilder.setRecord(narrowFunction.apply(builder.getRecord())); + return narrowBuilder.build(); + } + +}