abdullah alamoudi has uploaded a new change for review. https://asterix-gerrit.ics.uci.edu/2632
Change subject: WIP: Add last component id to index checkpoint files ...................................................................... WIP: Add last component id to index checkpoint files Change-Id: If24c9baaac2b79e7d1acf47fa2601767388ce988 --- M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IIndexCheckpointManager.java M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexCheckpoint.java M asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMIOOperationCallbackTest.java M asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/MarkComponentValidTask.java M asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/IndexSynchronizer.java M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeFlushOperation.java M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java 9 files changed, 78 insertions(+), 28 deletions(-) git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/32/2632/1 diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java index a012f1e..fbea0d0 100644 --- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java +++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java @@ -65,18 +65,20 @@ } @Override - public synchronized void replicated(String componentTimestamp, long masterLsn) throws HyracksDataException { + public synchronized void replicated(String componentTimestamp, long masterLsn, long componentId) + throws HyracksDataException { final Long localLsn = getLatest().getMasterNodeFlushMap().get(masterLsn); if (localLsn == null) { throw new IllegalStateException("Component flushed before lsn mapping was received"); } - flushed(componentTimestamp, localLsn); + flushed(componentTimestamp, localLsn, componentId); } @Override - public synchronized void flushed(String componentTimestamp, long lsn) throws HyracksDataException { + public synchronized void flushed(String componentTimestamp, long lsn, long componentId) + throws HyracksDataException { final IndexCheckpoint latest = getLatest(); - IndexCheckpoint nextCheckpoint = IndexCheckpoint.next(latest, lsn, componentTimestamp); + IndexCheckpoint nextCheckpoint = IndexCheckpoint.next(latest, lsn, componentTimestamp, componentId); persist(nextCheckpoint); deleteHistory(nextCheckpoint.getId(), HISTORY_CHECKPOINTS); } @@ -85,8 +87,8 @@ public synchronized void masterFlush(long masterLsn, long localLsn) throws HyracksDataException { final IndexCheckpoint latest = getLatest(); latest.getMasterNodeFlushMap().put(masterLsn, localLsn); - final IndexCheckpoint next = - IndexCheckpoint.next(latest, latest.getLowWatermark(), latest.getValidComponentTimestamp()); + final IndexCheckpoint next = IndexCheckpoint.next(latest, latest.getLowWatermark(), + latest.getValidComponentTimestamp(), latest.getLastComponentId()); persist(next); notifyAll(); } diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java index 50f5906..2674c4c 100644 --- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java +++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java @@ -23,7 +23,6 @@ import java.util.Deque; import java.util.List; import java.util.Map; -import java.util.Optional; import org.apache.asterix.common.context.DatasetInfo; import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider; @@ -45,6 +44,8 @@ import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexFileManager; import org.apache.hyracks.storage.am.lsm.common.impls.DiskComponentMetadata; import org.apache.hyracks.storage.am.lsm.common.impls.FlushOperation; +import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences; +import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentId; import org.apache.hyracks.storage.am.lsm.common.util.ComponentUtils; import org.apache.hyracks.storage.am.lsm.common.util.LSMComponentIdUtils; @@ -52,6 +53,7 @@ public class LSMIOOperationCallback implements ILSMIOOperationCallback { public static final String KEY_FLUSH_LOG_LSN = "FlushLogLsn"; public static final String KEY_NEXT_COMPONENT_ID = "NextComponentId"; + public static final String KEY_FLUSHED_COMPONENT_ID = "FlushedComponentId"; private static final String KEY_FIRST_LSN = "FirstLsn"; private static final MutableArrayValueReference KEY_METADATA_FLUSH_LOG_LSN = new MutableArrayValueReference(KEY_FLUSH_LOG_LSN.getBytes()); @@ -105,15 +107,16 @@ return; } if (operation.getIOOpertionType() == LSMIOOperationType.FLUSH) { + // will always update the checkpoint file even if no new component was created + FlushOperation flush = (FlushOperation) operation; + LSMComponentFileReferences files = flush.getComponentFiles(); Map<String, Object> map = operation.getAccessor().getOpContext().getParameters(); final Long lsn = (Long) map.get(KEY_FLUSH_LOG_LSN); - final Optional<String> componentFile = - operation.getNewComponent().getLSMComponentPhysicalFiles().stream().findAny(); - if (componentFile.isPresent()) { - final ResourceReference ref = ResourceReference.of(componentFile.get()); - final String componentEndTime = AbstractLSMIndexFileManager.getComponentEndTime(ref.getName()); - indexCheckpointManagerProvider.get(ref).flushed(componentEndTime, lsn); - } + final LSMComponentId id = (LSMComponentId) map.get(KEY_FLUSHED_COMPONENT_ID); + String componentFile = files.getInsertIndexFileReference().getAbsolutePath(); + final ResourceReference ref = ResourceReference.of(componentFile); + final String componentEndTime = AbstractLSMIndexFileManager.getComponentEndTime(ref.getName()); + indexCheckpointManagerProvider.get(ref).flushed(componentEndTime, lsn, id.getMinId()); } } @@ -195,9 +198,11 @@ dsInfo.declareActiveIOOperation(); if (operation.getIOOpertionType() == LSMIOOperationType.FLUSH) { pendingFlushes++; + FlushOperation flush = (FlushOperation) operation; Map<String, Object> map = operation.getAccessor().getOpContext().getParameters(); Long flushLsn = (Long) map.get(KEY_FLUSH_LOG_LSN); map.put(KEY_FIRST_LSN, firstLsnForCurrentMemoryComponent); + map.put(KEY_FLUSHED_COMPONENT_ID, flush.getFlushingComponent().getId()); componentIds.add((ILSMComponentId) map.get(KEY_NEXT_COMPONENT_ID)); firstLsnForCurrentMemoryComponent = flushLsn; // Advance the first lsn for new component } diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IIndexCheckpointManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IIndexCheckpointManager.java index b008f11..a172a0a 100644 --- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IIndexCheckpointManager.java +++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IIndexCheckpointManager.java @@ -33,25 +33,26 @@ void init(long lsn) throws HyracksDataException; /** - * Called when a new LSM disk component is flushed. When called, the index checkpoiint is updated + * Called when a new LSM disk component is flushed. When called, the index checkpoint is updated * with the latest valid {@code componentTimestamp} and low watermark {@code lsn} * * @param componentTimestamp * @param lsn * @throws HyracksDataException */ - void flushed(String componentTimestamp, long lsn) throws HyracksDataException; + void flushed(String componentTimestamp, long lsn, long componentId) throws HyracksDataException; /** - * Called when a new LSM disk component is replicated from master. When called, the index checkpoiint is updated + * Called when a new LSM disk component is replicated from master. When called, the index checkpoint is updated * with the latest valid {@code componentTimestamp} and the local lsn mapping of {@code masterLsn} is set as the * new low watermark. * * @param componentTimestamp * @param masterLsn + * @param componentId * @throws HyracksDataException */ - void replicated(String componentTimestamp, long masterLsn) throws HyracksDataException; + void replicated(String componentTimestamp, long masterLsn, long componentId) throws HyracksDataException; /** * Called when a flush log is received and replicated from master. The mapping between diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexCheckpoint.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexCheckpoint.java index 6e845e1..c12da9b 100644 --- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexCheckpoint.java +++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexCheckpoint.java @@ -32,9 +32,11 @@ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); private static final long INITIAL_CHECKPOINT_ID = 0; + private static final long EMPTY_INDEX_LAST_COMPONENT_ID = -1L; private long id; private String validComponentTimestamp; private long lowWatermark; + private long lastComponentId; private Map<Long, Long> masterNodeFlushMap; public static IndexCheckpoint first(long lowWatermark) { @@ -42,17 +44,20 @@ firstCheckpoint.id = INITIAL_CHECKPOINT_ID; firstCheckpoint.lowWatermark = lowWatermark; firstCheckpoint.validComponentTimestamp = null; + firstCheckpoint.lastComponentId = EMPTY_INDEX_LAST_COMPONENT_ID; firstCheckpoint.masterNodeFlushMap = new HashMap<>(); return firstCheckpoint; } - public static IndexCheckpoint next(IndexCheckpoint latest, long lowWatermark, String validComponentTimestamp) { + public static IndexCheckpoint next(IndexCheckpoint latest, long lowWatermark, String validComponentTimestamp, + long lastComponentId) { if (lowWatermark < latest.getLowWatermark()) { throw new IllegalStateException("Low watermark should always be increasing"); } IndexCheckpoint next = new IndexCheckpoint(); next.id = latest.getId() + 1; next.lowWatermark = lowWatermark; + next.lastComponentId = lastComponentId; next.validComponentTimestamp = validComponentTimestamp; next.masterNodeFlushMap = latest.getMasterNodeFlushMap(); // remove any lsn from the map that wont be used anymore @@ -72,6 +77,10 @@ return lowWatermark; } + public long getLastComponentId() { + return lastComponentId; + } + public Map<Long, Long> getMasterNodeFlushMap() { return masterNodeFlushMap; } diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMIOOperationCallbackTest.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMIOOperationCallbackTest.java index 7af7b6e..29a2aa0 100644 --- a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMIOOperationCallbackTest.java +++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMIOOperationCallbackTest.java @@ -19,6 +19,9 @@ package org.apache.asterix.test.ioopcallbacks; +import java.text.Format; +import java.text.SimpleDateFormat; +import java.util.Date; import java.util.HashMap; import java.util.Map; @@ -35,6 +38,7 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex; import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor; import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent; +import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexFileManager; import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMMemoryComponent; import org.apache.hyracks.storage.am.lsm.common.impls.DiskComponentMetadata; import org.apache.hyracks.storage.am.lsm.common.impls.FlushOperation; @@ -60,6 +64,15 @@ * 7. destroy */ + private static final Format FORMATTER = + new SimpleDateFormat(AbstractLSMIndexFileManager.COMPONENT_TIMESTAMP_FORMAT); + + private static String getComponentFileName() { + Date date = new Date(); + String ts = FORMATTER.format(date); + return ts + '_' + ts; + } + @Test public void testNormalSequence() throws HyracksDataException { int numMemoryComponents = 2; @@ -81,7 +94,7 @@ flushMap.put(LSMIOOperationCallback.KEY_NEXT_COMPONENT_ID, nextComponentId); ILSMIndexAccessor firstAccessor = new TestLSMIndexAccessor(new TestLSMIndexOperationContext(mockIndex)); firstAccessor.getOpContext().setParameters(flushMap); - FileReference firstTarget = new FileReference(Mockito.mock(IODeviceHandle.class), "path"); + FileReference firstTarget = new FileReference(Mockito.mock(IODeviceHandle.class), getComponentFileName()); LSMComponentFileReferences firstFiles = new LSMComponentFileReferences(firstTarget, firstTarget, firstTarget); FlushOperation firstFlush = new TestFlushOperation(firstAccessor, firstTarget, callback, indexId, firstFiles, new LSMComponentId(0, 0)); @@ -97,7 +110,7 @@ flushMap.put(LSMIOOperationCallback.KEY_NEXT_COMPONENT_ID, nextComponentId); ILSMIndexAccessor secondAccessor = new TestLSMIndexAccessor(new TestLSMIndexOperationContext(mockIndex)); secondAccessor.getOpContext().setParameters(flushMap); - FileReference secondTarget = new FileReference(Mockito.mock(IODeviceHandle.class), "path"); + FileReference secondTarget = new FileReference(Mockito.mock(IODeviceHandle.class), getComponentFileName()); LSMComponentFileReferences secondFiles = new LSMComponentFileReferences(secondTarget, secondTarget, secondTarget); FlushOperation secondFlush = new TestFlushOperation(secondAccessor, secondTarget, callback, indexId, @@ -175,7 +188,7 @@ flushMap.put(LSMIOOperationCallback.KEY_NEXT_COMPONENT_ID, expectedId); ILSMIndexAccessor accessor = new TestLSMIndexAccessor(new TestLSMIndexOperationContext(mockIndex)); accessor.getOpContext().setParameters(flushMap); - FileReference target = new FileReference(Mockito.mock(IODeviceHandle.class), "path"); + FileReference target = new FileReference(Mockito.mock(IODeviceHandle.class), getComponentFileName()); LSMComponentFileReferences files = new LSMComponentFileReferences(target, target, target); FlushOperation flush = new TestFlushOperation(accessor, target, callback, indexId, files, new LSMComponentId(0, 0)); @@ -210,7 +223,7 @@ IIndexCheckpointManagerProvider indexCheckpointManagerProvider = Mockito.mock(IIndexCheckpointManagerProvider.class); IIndexCheckpointManager indexCheckpointManager = Mockito.mock(IIndexCheckpointManager.class); - Mockito.doNothing().when(indexCheckpointManager).flushed(Mockito.any(), Mockito.anyLong()); + Mockito.doNothing().when(indexCheckpointManager).flushed(Mockito.any(), Mockito.anyLong(), Mockito.anyLong()); Mockito.doReturn(indexCheckpointManager).when(indexCheckpointManagerProvider).get(Mockito.any()); return indexCheckpointManagerProvider; } diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/MarkComponentValidTask.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/MarkComponentValidTask.java index 57474ef..7102988 100644 --- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/MarkComponentValidTask.java +++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/MarkComponentValidTask.java @@ -43,10 +43,12 @@ public class MarkComponentValidTask implements IReplicaTask { private final long masterLsn; + private final long lastComponentId; private final String file; - public MarkComponentValidTask(String file, long masterLsn) { + public MarkComponentValidTask(String file, long masterLsn, long lastComponentId) { this.file = file; + this.lastComponentId = lastComponentId; this.masterLsn = masterLsn; } @@ -82,7 +84,7 @@ replicationTimeOut -= TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime); } final String componentEndTime = AbstractLSMIndexFileManager.getComponentEndTime(indexRef.getName()); - indexCheckpointManager.replicated(componentEndTime, masterLsn); + indexCheckpointManager.replicated(componentEndTime, masterLsn, lastComponentId); } } @@ -97,6 +99,7 @@ final DataOutputStream dos = new DataOutputStream(out); dos.writeUTF(file); dos.writeLong(masterLsn); + dos.writeLong(lastComponentId); } catch (IOException e) { throw HyracksDataException.create(e); } @@ -105,6 +108,7 @@ public static MarkComponentValidTask create(DataInput input) throws IOException { final String indexFile = input.readUTF(); final long lsn = input.readLong(); - return new MarkComponentValidTask(indexFile, lsn); + final long lastComponentId = input.readLong(); + return new MarkComponentValidTask(indexFile, lsn, lastComponentId); } } \ No newline at end of file diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/IndexSynchronizer.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/IndexSynchronizer.java index 0e07ac7..502aeee 100644 --- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/IndexSynchronizer.java +++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/IndexSynchronizer.java @@ -37,6 +37,7 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext; import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexReplicationJob; import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType; +import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentId; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -91,7 +92,8 @@ final FileSynchronizer fileSynchronizer = new FileSynchronizer(appCtx, replica); job.getJobFiles().stream().map(StoragePathUtil::getFileRelativePath).forEach(fileSynchronizer::replicate); // send mark component valid - MarkComponentValidTask markValidTask = new MarkComponentValidTask(indexFile, getReplicatedComponentLsn()); + MarkComponentValidTask markValidTask = + new MarkComponentValidTask(indexFile, getReplicatedComponentLsn(), getReplicatedComponentId()); ReplicationProtocol.sendTo(replica, markValidTask); ReplicationProtocol.waitForAck(replica); LOGGER.debug("Replicated component ({}) to replica {}", indexFile, replica); @@ -126,4 +128,14 @@ return ((LSMIOOperationCallback) lsmIndex.getIOOperationCallback()) .getComponentLSN(ctx.getComponentsToBeReplicated()); } + + private long getReplicatedComponentId() throws HyracksDataException { + final ILSMIndexReplicationJob indexReplJob = (ILSMIndexReplicationJob) job; + if (indexReplJob.getLSMOpType() != LSMOperationType.FLUSH) { + return -1L; + } + final ILSMIndexOperationContext ctx = indexReplJob.getLSMIndexOperationContext(); + LSMComponentId id = (LSMComponentId) ctx.getComponentsToBeReplicated().get(0).getId(); + return id.getMinId(); + } } \ No newline at end of file diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeFlushOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeFlushOperation.java index 7acc59f..2ad4fee 100644 --- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeFlushOperation.java +++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeFlushOperation.java @@ -26,11 +26,13 @@ public class LSMBTreeFlushOperation extends FlushOperation { private final FileReference bloomFilterFlushTarget; + private final LSMComponentFileReferences fileReferences; public LSMBTreeFlushOperation(ILSMIndexAccessor accessor, FileReference flushTarget, FileReference bloomFilterFlushTarget, ILSMIOOperationCallback callback, String indexIdentifier) { super(accessor, flushTarget, callback, indexIdentifier); this.bloomFilterFlushTarget = bloomFilterFlushTarget; + fileReferences = new LSMComponentFileReferences(target, null, bloomFilterFlushTarget); } public FileReference getBloomFilterTarget() { @@ -39,6 +41,6 @@ @Override public LSMComponentFileReferences getComponentFiles() { - return new LSMComponentFileReferences(target, null, bloomFilterFlushTarget); + return fileReferences; } } diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java index 2a3e983..ece7688 100644 --- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java +++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java @@ -271,6 +271,8 @@ // newComponent is null if the flush op. was not performed. if (!failedOperation && newComponent != null) { lsmIndex.addDiskComponent(newComponent); + // TODO: The following should also replicate component Id + // even if empty component if (replicationEnabled && newComponent != EmptyComponent.INSTANCE) { componentsToBeReplicated.clear(); componentsToBeReplicated.add(newComponent); -- To view, visit https://asterix-gerrit.ics.uci.edu/2632 To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings Gerrit-MessageType: newchange Gerrit-Change-Id: If24c9baaac2b79e7d1acf47fa2601767388ce988 Gerrit-PatchSet: 1 Gerrit-Project: asterixdb Gerrit-Branch: master Gerrit-Owner: abdullah alamoudi <bamou...@gmail.com>