This is an automated email from the ASF dual-hosted git repository.
wyk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
The following commit(s) were added to refs/heads/master by this push:
new a267a6bda5 [ASTERIXDB-3438][STO] Ensure masks are written to cloud
a267a6bda5 is described below
commit a267a6bda50698593444882b4bf1c2ad478d280f
Author: Wail Alkowaileet <[email protected]>
AuthorDate: Thu Jun 20 07:17:17 2024 -0700
[ASTERIXDB-3438][STO] Ensure masks are written to cloud
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
Masks are not written to cloud, which could lead to
consider uncompleted merged components as valid.
Change-Id: I13afb363de3fe6dafabffd0d72c3cb037780f633
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/18387
Integration-Tests: Jenkins <[email protected]>
Tested-by: Jenkins <[email protected]>
Reviewed-by: Wail Alkowaileet <[email protected]>
Reviewed-by: Murtadha Hubail <[email protected]>
---
.../apache/asterix/app/nc/task/MetadataBootstrapTask.java | 2 +-
.../asterix/test/dataflow/TestLsmIoOpCallbackFactory.java | 7 ++++---
.../org/apache/asterix/cloud/AbstractCloudIOManager.java | 2 ++
.../common/ioopcallbacks/AtomicLSMIOOperationCallback.java | 8 ++++----
.../AtomicLSMIndexIOOperationCallbackFactory.java | 3 ++-
.../common/ioopcallbacks/LSMIOOperationCallback.java | 14 ++++++++------
.../ioopcallbacks/LSMIndexIOOperationCallbackFactory.java | 7 ++++++-
.../test/ioopcallbacks/LSMIOOperationCallbackTest.java | 11 ++++++++---
8 files changed, 35 insertions(+), 19 deletions(-)
diff --git
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/task/MetadataBootstrapTask.java
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/task/MetadataBootstrapTask.java
index 246983de78..e4f5ed97d8 100644
---
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/task/MetadataBootstrapTask.java
+++
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/task/MetadataBootstrapTask.java
@@ -60,7 +60,7 @@ public class MetadataBootstrapTask implements
INCLifecycleTask {
IIOManager persistenceIoManager = appContext.getPersistenceIoManager();
FileReference bootstrapMarker = persistenceIoManager
.resolve(StoragePathUtil.getBootstrapMarkerRelativePath(appContext.getNamespacePathResolver()));
- persistenceIoManager.overwrite(bootstrapMarker, new byte[0]);
+ persistenceIoManager.create(bootstrapMarker);
}
private void deleteBootstrapMarker(INcApplicationContext appContext)
throws HyracksDataException {
diff --git
a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmIoOpCallbackFactory.java
b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmIoOpCallbackFactory.java
index 0d8677f157..fa25f6d112 100644
---
a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmIoOpCallbackFactory.java
+++
b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmIoOpCallbackFactory.java
@@ -25,6 +25,7 @@ import
org.apache.asterix.common.ioopcallbacks.LSMIOOperationCallback;
import
org.apache.asterix.common.ioopcallbacks.LSMIndexIOOperationCallbackFactory;
import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.api.io.IJsonSerializable;
import org.apache.hyracks.api.io.IPersistedResourceRegistry;
import org.apache.hyracks.storage.am.lsm.btree.impl.TestLsmBtree;
@@ -65,7 +66,7 @@ public class TestLsmIoOpCallbackFactory extends
LSMIndexIOOperationCallbackFacto
// However, the counters for the failed operations are never reset
since we expect them
// To be always 0
return new
TestLsmIoOpCallback(datasetInfoProvider.getDatasetInfo(ncCtx), index,
getComponentIdGenerator(),
- getIndexCheckpointManagerProvider());
+ getIndexCheckpointManagerProvider(), getIOManager());
}
public int getTotalFlushes() {
@@ -126,8 +127,8 @@ public class TestLsmIoOpCallbackFactory extends
LSMIndexIOOperationCallbackFacto
private final TestLsmBtree lsmBtree;
public TestLsmIoOpCallback(DatasetInfo dsInfo, ILSMIndex index,
ILSMComponentIdGenerator idGenerator,
- IIndexCheckpointManagerProvider checkpointManagerProvider) {
- super(dsInfo, index, idGenerator.getId(),
checkpointManagerProvider);
+ IIndexCheckpointManagerProvider checkpointManagerProvider,
IIOManager ioManager) {
+ super(dsInfo, index, idGenerator.getId(),
checkpointManagerProvider, ioManager);
lsmBtree = (TestLsmBtree) index;
}
diff --git
a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/AbstractCloudIOManager.java
b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/AbstractCloudIOManager.java
index c0823af567..4dfcd5f2d0 100644
---
a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/AbstractCloudIOManager.java
+++
b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/AbstractCloudIOManager.java
@@ -60,6 +60,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
public abstract class AbstractCloudIOManager extends IOManager implements
IPartitionBootstrapper, ICloudIOManager {
private static final Logger LOGGER = LogManager.getLogger();
+ private static final byte[] EMPTY_FILE_BYTES = "empty".getBytes();
protected final ICloudClient cloudClient;
protected final ICloudGuardian guardian;
protected final IWriteBufferProvider writeBufferProvider;
@@ -302,6 +303,7 @@ public abstract class AbstractCloudIOManager extends
IOManager implements IParti
public final void create(FileReference fileRef) throws
HyracksDataException {
// We need to delete the local file on create as the cloud storage
didn't complete the upload
// In other words, both cloud files and the local files are not in sync
+ overwrite(fileRef, EMPTY_FILE_BYTES);
localIoManager.delete(fileRef);
localIoManager.create(fileRef);
}
diff --git
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AtomicLSMIOOperationCallback.java
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AtomicLSMIOOperationCallback.java
index 08273a745a..9ac0410497 100644
---
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AtomicLSMIOOperationCallback.java
+++
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AtomicLSMIOOperationCallback.java
@@ -22,7 +22,7 @@ package org.apache.asterix.common.ioopcallbacks;
import org.apache.asterix.common.context.DatasetInfo;
import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.api.util.IoUtil;
+import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
@@ -33,8 +33,8 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
public class AtomicLSMIOOperationCallback extends LSMIOOperationCallback {
public AtomicLSMIOOperationCallback(DatasetInfo dsInfo, ILSMIndex
lsmIndex, ILSMComponentId componentId,
- IIndexCheckpointManagerProvider indexCheckpointManagerProvider) {
- super(dsInfo, lsmIndex, componentId, indexCheckpointManagerProvider);
+ IIndexCheckpointManagerProvider indexCheckpointManagerProvider,
IIOManager ioManager) {
+ super(dsInfo, lsmIndex, componentId, indexCheckpointManagerProvider,
ioManager);
}
@Override
@@ -48,7 +48,7 @@ public class AtomicLSMIOOperationCallback extends
LSMIOOperationCallback {
} else if (operation.getIOOperationType() == LSMIOOperationType.LOAD) {
addComponentToCheckpoint(operation);
} else if (isMerge(operation)) {
- IoUtil.delete(getOperationMaskFilePath(operation));
+ ioManager.delete(getOperationMaskFilePath(operation));
}
}
}
diff --git
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AtomicLSMIndexIOOperationCallbackFactory.java
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AtomicLSMIndexIOOperationCallbackFactory.java
index 3829c547cd..259235b11a 100644
---
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AtomicLSMIndexIOOperationCallbackFactory.java
+++
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AtomicLSMIndexIOOperationCallbackFactory.java
@@ -40,6 +40,7 @@ public class AtomicLSMIndexIOOperationCallbackFactory extends
LSMIndexIOOperatio
super(idGeneratorFactory, datasetInfoProvider);
}
+ @Override
protected IIndexCheckpointManagerProvider
getIndexCheckpointManagerProvider() {
return ((INcApplicationContext)
ncCtx.getApplicationContext()).getIndexCheckpointManagerProvider();
}
@@ -47,7 +48,7 @@ public class AtomicLSMIndexIOOperationCallbackFactory extends
LSMIndexIOOperatio
@Override
public ILSMIOOperationCallback createIoOpCallback(ILSMIndex index) throws
HyracksDataException {
return new
AtomicLSMIOOperationCallback(datasetInfoProvider.getDatasetInfo(ncCtx), index,
- getComponentIdGenerator().getId(),
getIndexCheckpointManagerProvider());
+ getComponentIdGenerator().getId(),
getIndexCheckpointManagerProvider(), getIOManager());
}
@SuppressWarnings("squid:S1172") // unused parameter
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 fc895bc021..87aa3bd3c0 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
@@ -33,7 +33,7 @@ import org.apache.asterix.common.storage.ResourceReference;
import org.apache.asterix.common.utils.StorageConstants;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
-import org.apache.hyracks.api.util.IoUtil;
+import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.data.std.primitive.LongPointable;
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
@@ -69,6 +69,7 @@ public class LSMIOOperationCallback implements
ILSMIOOperationCallback {
public static final long INVALID_LSN = -1L;
private final ArrayBackedValueStorage buffer = new
ArrayBackedValueStorage(Long.BYTES);
private final IIndexCheckpointManagerProvider
indexCheckpointManagerProvider;
+ protected final IIOManager ioManager;
protected final DatasetInfo dsInfo;
protected final ILSMIndex lsmIndex;
private final int partition;
@@ -78,11 +79,12 @@ public class LSMIOOperationCallback implements
ILSMIOOperationCallback {
private final Deque<ILSMComponentId> componentIds = new ArrayDeque<>();
public LSMIOOperationCallback(DatasetInfo dsInfo, ILSMIndex lsmIndex,
ILSMComponentId componentId,
- IIndexCheckpointManagerProvider indexCheckpointManagerProvider) {
+ IIndexCheckpointManagerProvider indexCheckpointManagerProvider,
IIOManager ioManager) {
this.dsInfo = dsInfo;
this.lsmIndex = lsmIndex;
this.indexCheckpointManagerProvider = indexCheckpointManagerProvider;
this.partition =
ResourceReference.ofIndex(lsmIndex.getIndexIdentifier()).getPartitionNum();
+ this.ioManager = ioManager;
componentIds.add(componentId);
}
@@ -90,9 +92,9 @@ public class LSMIOOperationCallback implements
ILSMIOOperationCallback {
public void beforeOperation(ILSMIOOperation operation) throws
HyracksDataException {
if (isMerge(operation)) {
FileReference operationMaskFilePath =
getOperationMaskFilePath(operation);
- // if a merge operation is attempted after a failure, its mask
file may already exists
- if (!operationMaskFilePath.getFile().exists()) {
- IoUtil.create(operationMaskFilePath);
+ // if a merge operation is attempted after a failure, its mask
file may already exist
+ if (!ioManager.exists(operationMaskFilePath)) {
+ ioManager.create(operationMaskFilePath);
} else {
LOGGER.warn("merge operation mask file {} already exists",
operationMaskFilePath);
}
@@ -135,7 +137,7 @@ public class LSMIOOperationCallback implements
ILSMIOOperationCallback {
|| operation.getIOOperationType() == LSMIOOperationType.LOAD) {
addComponentToCheckpoint(operation);
} else if (isMerge(operation)) {
- IoUtil.delete(getOperationMaskFilePath(operation));
+ ioManager.delete(getOperationMaskFilePath(operation));
}
}
diff --git
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIndexIOOperationCallbackFactory.java
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIndexIOOperationCallbackFactory.java
index 25cd8b2681..2bb4c89b6e 100644
---
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIndexIOOperationCallbackFactory.java
+++
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIndexIOOperationCallbackFactory.java
@@ -25,6 +25,7 @@ import org.apache.asterix.common.api.INcApplicationContext;
import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
import org.apache.hyracks.api.application.INCServiceContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.api.io.IJsonSerializable;
import org.apache.hyracks.api.io.IPersistedResourceRegistry;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
@@ -68,10 +69,14 @@ public class LSMIndexIOOperationCallbackFactory implements
ILSMIOOperationCallba
return ((INcApplicationContext)
ncCtx.getApplicationContext()).getIndexCheckpointManagerProvider();
}
+ protected IIOManager getIOManager() {
+ return ((INcApplicationContext)
ncCtx.getApplicationContext()).getPersistenceIoManager();
+ }
+
@Override
public ILSMIOOperationCallback createIoOpCallback(ILSMIndex index) throws
HyracksDataException {
return new
LSMIOOperationCallback(datasetInfoProvider.getDatasetInfo(ncCtx), index,
- getComponentIdGenerator().getId(),
getIndexCheckpointManagerProvider());
+ getComponentIdGenerator().getId(),
getIndexCheckpointManagerProvider(), getIOManager());
}
@Override
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 befdd1a915..99636596a4 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
@@ -31,6 +31,7 @@ import
org.apache.asterix.common.storage.IIndexCheckpointManager;
import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.api.io.IODeviceHandle;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
@@ -82,7 +83,7 @@ public class LSMIOOperationCallbackTest extends TestCase {
DatasetInfo dsInfo = new DatasetInfo(101, null);
LSMComponentIdGenerator idGenerator = new
LSMComponentIdGenerator(numMemoryComponents, MIN_VALID_COMPONENT_ID);
LSMIOOperationCallback callback = new LSMIOOperationCallback(dsInfo,
mockIndex, idGenerator.getId(),
- mockIndexCheckpointManagerProvider());
+ mockIndexCheckpointManagerProvider(), mockIOManager());
//Flush first
idGenerator.refresh();
long flushLsn = 1L;
@@ -148,7 +149,7 @@ public class LSMIOOperationCallbackTest extends TestCase {
ILSMMemoryComponent mockComponent =
Mockito.mock(AbstractLSMMemoryComponent.class);
Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(mockComponent);
LSMIOOperationCallback callback = new LSMIOOperationCallback(dsInfo,
mockIndex, idGenerator.getId(),
- mockIndexCheckpointManagerProvider());
+ mockIndexCheckpointManagerProvider(), mockIOManager());
ILSMComponentId initialId = idGenerator.getId();
// simulate a partition is flushed before allocated
idGenerator.refresh();
@@ -170,7 +171,7 @@ public class LSMIOOperationCallbackTest extends TestCase {
ILSMMemoryComponent mockComponent =
Mockito.mock(AbstractLSMMemoryComponent.class);
Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(mockComponent);
LSMIOOperationCallback callback = new LSMIOOperationCallback(dsInfo,
mockIndex, idGenerator.getId(),
- mockIndexCheckpointManagerProvider());
+ mockIndexCheckpointManagerProvider(), mockIOManager());
String indexId = "mockIndexId";
ILSMComponentId id = idGenerator.getId();
callback.recycled(mockComponent);
@@ -230,6 +231,10 @@ public class LSMIOOperationCallbackTest extends TestCase {
return indexCheckpointManagerProvider;
}
+ private IIOManager mockIOManager() {
+ return Mockito.mock(IIOManager.class);
+ }
+
private static String getIndexPath() {
return "storage/partition_0/dataverse/dataset/0/index";
}
