You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by mh...@apache.org on 2018/04/30 20:50:42 UTC
[7/8] asterixdb git commit: [NO ISSUE][STO] Add consistency to flush
lifecycle
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicy.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicy.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicy.java
index 41461ec..a14fac8 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicy.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicy.java
@@ -125,7 +125,7 @@ public class CorrelatedPrefixMergePolicy extends PrefixMergePolicy {
}
}
ILSMIndexAccessor accessor = lsmIndex.createAccessor(NoOpIndexAccessParameters.INSTANCE);
- accessor.scheduleMerge(lsmIndex.getIOOperationCallback(), mergableComponents);
+ accessor.scheduleMerge(mergableComponents);
}
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfoProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfoProvider.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfoProvider.java
new file mode 100644
index 0000000..597b5da
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfoProvider.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.asterix.common.context;
+
+import org.apache.asterix.common.api.IDatasetInfoProvider;
+import org.apache.asterix.common.api.IDatasetLifecycleManager;
+import org.apache.asterix.common.api.INcApplicationContext;
+import org.apache.hyracks.api.application.INCServiceContext;
+
+public class DatasetInfoProvider implements IDatasetInfoProvider {
+
+ private static final long serialVersionUID = 1L;
+ private final int datasetId;
+
+ public DatasetInfoProvider(int datasetId) {
+ this.datasetId = datasetId;
+ }
+
+ @Override
+ public DatasetInfo getDatasetInfo(INCServiceContext serviceCtx) {
+ IDatasetLifecycleManager dslcManager =
+ ((INcApplicationContext) serviceCtx.getApplicationContext()).getDatasetLifecycleManager();
+ return dslcManager.getDatasetInfo(datasetId);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLSMComponentIdGeneratorFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLSMComponentIdGeneratorFactory.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLSMComponentIdGeneratorFactory.java
index 83e3144..85eacf8 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLSMComponentIdGeneratorFactory.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLSMComponentIdGeneratorFactory.java
@@ -20,12 +20,12 @@
package org.apache.asterix.common.context;
import org.apache.asterix.common.api.IDatasetLifecycleManager;
+import org.apache.asterix.common.api.ILSMComponentIdGeneratorFactory;
import org.apache.asterix.common.api.INcApplicationContext;
import org.apache.asterix.common.utils.StoragePathUtil;
import org.apache.hyracks.api.application.INCServiceContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGeneratorFactory;
import org.apache.hyracks.storage.common.IResource;
/**
@@ -51,5 +51,4 @@ public class DatasetLSMComponentIdGeneratorFactory implements ILSMComponentIdGen
int partition = StoragePathUtil.getPartitionNumFromRelativePath(resource.getPath());
return dslcManager.getComponentIdGenerator(datasetId, partition);
}
-
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
index b715eec..e5d18cf 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
@@ -25,6 +25,7 @@ import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -34,7 +35,7 @@ import org.apache.asterix.common.api.IDatasetMemoryManager;
import org.apache.asterix.common.config.StorageProperties;
import org.apache.asterix.common.dataflow.DatasetLocalResource;
import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
+import org.apache.asterix.common.ioopcallbacks.LSMIOOperationCallback;
import org.apache.asterix.common.metadata.MetadataIndexImmutableProperties;
import org.apache.asterix.common.replication.IReplicationStrategy;
import org.apache.asterix.common.storage.DatasetResourceReference;
@@ -47,6 +48,7 @@ import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.lifecycle.ILifeCycleComponent;
import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
@@ -348,7 +350,7 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
}
private void populateOpTrackerAndIdGenerator(DatasetResource dataset, int partition) {
- ILSMComponentIdGenerator idGenerator = new LSMComponentIdGenerator();
+ ILSMComponentIdGenerator idGenerator = new LSMComponentIdGenerator(storageProperties.getMemoryComponentsNum());
PrimaryIndexOperationTracker opTracker = new PrimaryIndexOperationTracker(dataset.getDatasetID(), partition,
logManager, dataset.getDatasetInfo(), idGenerator);
dataset.setPrimaryIndexOperationTracker(partition, opTracker);
@@ -398,11 +400,10 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
PrimaryIndexOperationTracker opTracker, long targetLSN) throws HyracksDataException {
int partition = opTracker.getPartition();
for (ILSMIndex lsmIndex : dsr.getDatasetInfo().getDatasetPartitionOpenIndexes(partition)) {
- AbstractLSMIOOperationCallback ioCallback =
- (AbstractLSMIOOperationCallback) lsmIndex.getIOOperationCallback();
+ LSMIOOperationCallback ioCallback = (LSMIOOperationCallback) lsmIndex.getIOOperationCallback();
if (!(lsmIndex.isCurrentMutableComponentEmpty() || ioCallback.hasPendingFlush()
|| opTracker.isFlushLogCreated() || opTracker.isFlushOnExit())) {
- long firstLSN = ioCallback.getFirstLSN();
+ long firstLSN = ioCallback.getPersistenceLsn();
if (firstLSN < targetLSN) {
LOGGER.info("Checkpoint flush dataset {} partition {}", dsr.getDatasetID(), partition);
opTracker.setFlushOnExit(true);
@@ -448,9 +449,7 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
LSMComponentId componentId = (LSMComponentId) flushIndex.getCurrentMemoryComponent().getId();
ILSMComponentIdGenerator idGenerator = getComponentIdGenerator(dsInfo.getDatasetID(), partition);
idGenerator.refresh();
-
if (dsInfo.isDurable()) {
-
synchronized (logRecord) {
TransactionUtil.formFlushLogRecord(logRecord, dsInfo.getDatasetID(), partition,
componentId.getMinId(), componentId.getMaxId(), null);
@@ -469,32 +468,21 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
}
}
}
+ long flushLsn = logRecord.getLSN();
+ ILSMComponentId nextComponentId = idGenerator.getId();
+ Map<String, Object> flushMap = new HashMap<>();
+ flushMap.put(LSMIOOperationCallback.KEY_FLUSH_LOG_LSN, flushLsn);
+ flushMap.put(LSMIOOperationCallback.KEY_NEXT_COMPONENT_ID, nextComponentId);
for (ILSMIndex index : indexes) {
- //update resource lsn
- AbstractLSMIOOperationCallback ioOpCallback =
- (AbstractLSMIOOperationCallback) index.getIOOperationCallback();
- ioOpCallback.updateLastLSN(logRecord.getLSN());
+ ILSMIndexAccessor accessor = index.createAccessor(NoOpIndexAccessParameters.INSTANCE);
+ accessor.getOpContext().setParameters(flushMap);
+ accessor.scheduleFlush();
}
-
- if (asyncFlush) {
- for (ILSMIndex index : indexes) {
- ILSMIndexAccessor accessor = index.createAccessor(NoOpIndexAccessParameters.INSTANCE);
- accessor.scheduleFlush(index.getIOOperationCallback());
- }
- } else {
- for (ILSMIndex index : indexes) {
- // TODO: This is not efficient since we flush the indexes sequentially.
- // Think of a way to allow submitting the flush requests concurrently.
- // We don't do them concurrently because this may lead to a deadlock scenario
- // between the DatasetLifeCycleManager and the PrimaryIndexOperationTracker.
- ILSMIndexAccessor accessor = index.createAccessor(NoOpIndexAccessParameters.INSTANCE);
- accessor.scheduleFlush(index.getIOOperationCallback());
- // Wait for the above flush op.
- dsInfo.waitForIO();
- }
+ if (!asyncFlush) {
+ // Wait for the above flush op.
+ dsInfo.waitForIO();
}
}
-
}
private void closeDataset(DatasetResource dsr) throws HyracksDataException {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
index 74848d5..681669f 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
@@ -19,11 +19,13 @@
package org.apache.asterix.common.context;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
+import org.apache.asterix.common.ioopcallbacks.LSMIOOperationCallback;
import org.apache.asterix.common.metadata.MetadataIndexImmutableProperties;
import org.apache.asterix.common.transactions.AbstractOperationCallback;
import org.apache.asterix.common.transactions.ILogManager;
@@ -33,6 +35,7 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent.ComponentState;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
@@ -44,7 +47,6 @@ import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
public class PrimaryIndexOperationTracker extends BaseOperationTracker {
-
private final int partition;
// Number of active operations on an ILSMIndex instance.
private final AtomicInteger numActiveOperations;
@@ -65,20 +67,9 @@ public class PrimaryIndexOperationTracker extends BaseOperationTracker {
@Override
public void beforeOperation(ILSMIndex index, LSMOperationType opType, ISearchOperationCallback searchCallback,
IModificationOperationCallback modificationCallback) throws HyracksDataException {
+ super.beforeOperation(index, opType, searchCallback, modificationCallback);
if (opType == LSMOperationType.MODIFICATION || opType == LSMOperationType.FORCE_MODIFICATION) {
incrementNumActiveOperations(modificationCallback);
- } else if (opType == LSMOperationType.FLUSH || opType == LSMOperationType.MERGE
- || opType == LSMOperationType.REPLICATE) {
- dsInfo.declareActiveIOOperation();
- }
- }
-
- @Override
- public void afterOperation(ILSMIndex index, LSMOperationType opType, ISearchOperationCallback searchCallback,
- IModificationOperationCallback modificationCallback) throws HyracksDataException {
- // Searches are immediately considered complete, because they should not prevent the execution of flushes.
- if (opType == LSMOperationType.REPLICATE) {
- completeOperation(index, opType, searchCallback, modificationCallback);
}
}
@@ -86,12 +77,10 @@ public class PrimaryIndexOperationTracker extends BaseOperationTracker {
public synchronized void completeOperation(ILSMIndex index, LSMOperationType opType,
ISearchOperationCallback searchCallback, IModificationOperationCallback modificationCallback)
throws HyracksDataException {
+ super.completeOperation(index, opType, searchCallback, modificationCallback);
if (opType == LSMOperationType.MODIFICATION || opType == LSMOperationType.FORCE_MODIFICATION) {
decrementNumActiveOperations(modificationCallback);
flushIfNeeded();
- } else if (opType == LSMOperationType.FLUSH || opType == LSMOperationType.MERGE
- || opType == LSMOperationType.REPLICATE) {
- dsInfo.undeclareActiveIOOperation();
}
}
@@ -121,12 +110,29 @@ public class PrimaryIndexOperationTracker extends BaseOperationTracker {
// make the current mutable components READABLE_UNWRITABLE to stop coming modify operations from entering
// them until the current flush is scheduled.
LSMComponentId primaryId = null;
+ //Double check that the primary index has been modified
+ synchronized (this) {
+ if (numActiveOperations.get() > 0) {
+ throw new IllegalStateException(
+ "Can't request a flush on an index with active operations: " + numActiveOperations.get());
+ }
+ for (ILSMIndex lsmIndex : indexes) {
+ if (lsmIndex.isPrimaryIndex() && lsmIndex.isCurrentMutableComponentEmpty()) {
+ return;
+ }
+ }
+ }
for (ILSMIndex lsmIndex : indexes) {
ILSMOperationTracker opTracker = lsmIndex.getOperationTracker();
synchronized (opTracker) {
ILSMMemoryComponent memComponent = lsmIndex.getCurrentMemoryComponent();
+ if (memComponent.getWriterCount() > 0) {
+ throw new IllegalStateException(
+ "Can't request a flush on a component with writers inside: Index:" + lsmIndex
+ + " Component:" + memComponent);
+ }
if (memComponent.getState() == ComponentState.READABLE_WRITABLE && memComponent.isModified()) {
- memComponent.setState(ComponentState.READABLE_UNWRITABLE);
+ memComponent.setUnwritable();
}
if (lsmIndex.isPrimaryIndex()) {
primaryId = (LSMComponentId) memComponent.getId();
@@ -171,15 +177,15 @@ public class PrimaryIndexOperationTracker extends BaseOperationTracker {
throw new IllegalStateException("Operation started while index was pending scheduling a flush");
}
idGenerator.refresh();
+ long flushLsn = logRecord.getLSN();
+ ILSMComponentId nextComponentId = idGenerator.getId();
+ Map<String, Object> flushMap = new HashMap<>();
+ flushMap.put(LSMIOOperationCallback.KEY_FLUSH_LOG_LSN, flushLsn);
+ flushMap.put(LSMIOOperationCallback.KEY_NEXT_COMPONENT_ID, nextComponentId);
for (ILSMIndex lsmIndex : dsInfo.getDatasetPartitionOpenIndexes(partition)) {
- //get resource
ILSMIndexAccessor accessor = lsmIndex.createAccessor(NoOpIndexAccessParameters.INSTANCE);
- //update resource lsn
- AbstractLSMIOOperationCallback ioOpCallback =
- (AbstractLSMIOOperationCallback) lsmIndex.getIOOperationCallback();
- ioOpCallback.updateLastLSN(logRecord.getLSN());
- //schedule flush after update
- accessor.scheduleFlush(lsmIndex.getIOOperationCallback());
+ accessor.getOpContext().setParameters(flushMap);
+ accessor.scheduleFlush();
}
} finally {
flushLogCreated = false;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMIndexUtil.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMIndexUtil.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMIndexUtil.java
index e844192..99ab2d0 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMIndexUtil.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMIndexUtil.java
@@ -18,15 +18,16 @@
*/
package org.apache.asterix.common.dataflow;
-import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
+import org.apache.asterix.common.ioopcallbacks.LSMIOOperationCallback;
import org.apache.asterix.common.transactions.ILogManager;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
public class LSMIndexUtil {
+ private LSMIndexUtil() {
+ }
+
public static void checkAndSetFirstLSN(AbstractLSMIndex lsmIndex, ILogManager logManager)
throws HyracksDataException {
// If the index has an empty memory component, we need to set its first LSN (For soft checkpoint)
@@ -34,9 +35,8 @@ public class LSMIndexUtil {
//prevent transactions from incorrectly setting the first LSN on a modified component by checking the index is still empty
synchronized (lsmIndex.getOperationTracker()) {
if (lsmIndex.isCurrentMutableComponentEmpty()) {
- AbstractLSMIOOperationCallback ioOpCallback =
- (AbstractLSMIOOperationCallback) lsmIndex.getIOOperationCallback();
- ioOpCallback.setFirstLSN(logManager.getAppendLSN());
+ LSMIOOperationCallback ioOpCallback = (LSMIOOperationCallback) lsmIndex.getIOOperationCallback();
+ ioOpCallback.setFirstLsnForCurrentMemoryComponent(logManager.getAppendLSN());
}
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ExceptionUtils.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ExceptionUtils.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ExceptionUtils.java
index 256ce08..359054e 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ExceptionUtils.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ExceptionUtils.java
@@ -20,6 +20,8 @@ package org.apache.asterix.common.exceptions;
import java.util.function.Predicate;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
public class ExceptionUtils {
public static final String INCORRECT_PARAMETER = "Incorrect parameter.\n";
public static final String PARAMETER_NAME = "Parameter name: ";
@@ -45,13 +47,28 @@ public class ExceptionUtils {
Throwable current = e;
Throwable cause = e.getCause();
while (cause != null && cause != current) {
- Throwable nextCause = current.getCause();
current = cause;
- cause = nextCause;
+ cause = current.getCause();
}
return current;
}
+ public static Throwable getCause(Throwable e, String component, int code) {
+ Throwable current = e;
+ Throwable expected =
+ (current instanceof HyracksDataException && ((HyracksDataException) current).getErrorCode() == code
+ && ((HyracksDataException) current).getComponent().equals(component)) ? current : null;
+ Throwable cause = e.getCause();
+ while (cause != null && cause != current) {
+ current = cause;
+ expected =
+ (current instanceof HyracksDataException && ((HyracksDataException) current).getErrorCode() == code
+ && ((HyracksDataException) current).getComponent().equals(component)) ? current : expected;
+ cause = current.getCause();
+ }
+ return expected == null ? current : expected;
+ }
+
/**
* Determines whether supplied exception contains a matching cause in its hierarchy, or is itself a match
*/
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
deleted file mode 100644
index f027979..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
+++ /dev/null
@@ -1,287 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.asterix.common.ioopcallbacks;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
-import org.apache.asterix.common.storage.ResourceReference;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-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;
-import org.apache.hyracks.storage.am.common.freepage.MutableArrayValueReference;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationType;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
-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.DiskComponentMetadata;
-import org.apache.hyracks.storage.am.lsm.common.util.ComponentUtils;
-import org.apache.hyracks.storage.am.lsm.common.util.LSMComponentIdUtils;
-
-// A single LSMIOOperationCallback per LSM index used to perform actions around Flush and Merge operations
-public abstract class AbstractLSMIOOperationCallback implements ILSMIOOperationCallback {
- public static final MutableArrayValueReference LSN_KEY = new MutableArrayValueReference("LSN".getBytes());
- public static final long INVALID = -1L;
-
- protected final ILSMIndex lsmIndex;
- // First LSN per mutable component. TODO: move from hyracks to asterixdb
- protected final long[] firstLSNs;
- // A boolean array to keep track of flush operations
- protected final boolean[] flushRequested;
- // TODO: move from hyracks to asterixdb
- protected final long[] mutableLastLSNs;
- // Index of the currently flushing or next to be flushed component
- protected int readIndex;
- // Index of the currently being written to component
- protected int writeIndex;
- // Index of the memory component to be recycled
- protected int recycleIndex;
- // Indicates whether this index has been scheduled to flush (no matter whether succeeds or not)
- protected boolean hasFlushed;
- // Keep track of the component Id of the next component being activated.
- protected ILSMComponentId[] nextComponentIds;
-
- protected final ILSMComponentIdGenerator idGenerator;
- protected final ArrayBackedValueStorage buffer = new ArrayBackedValueStorage(Long.BYTES);
- private final IIndexCheckpointManagerProvider indexCheckpointManagerProvider;
- private final Map<ILSMComponentId, Long> componentLsnMap = new HashMap<>();
-
- public AbstractLSMIOOperationCallback(ILSMIndex lsmIndex, ILSMComponentIdGenerator idGenerator,
- IIndexCheckpointManagerProvider indexCheckpointManagerProvider) {
- this.lsmIndex = lsmIndex;
- this.idGenerator = idGenerator;
- this.indexCheckpointManagerProvider = indexCheckpointManagerProvider;
- int count = lsmIndex.getNumberOfAllMemoryComponents();
- mutableLastLSNs = new long[count];
- firstLSNs = new long[count];
- flushRequested = new boolean[count];
- readIndex = 0;
- writeIndex = 0;
- recycleIndex = 0;
- hasFlushed = false;
- nextComponentIds = new ILSMComponentId[count];
- if (count > 0) {
- nextComponentIds[0] = idGenerator.getId();
- }
- }
-
- @Override
- public void beforeOperation(ILSMIndexOperationContext opCtx) throws HyracksDataException {
- if (opCtx.getIoOperationType() == LSMIOOperationType.FLUSH) {
- /*
- * This method was called on the scheduleFlush operation.
- * We set the lastLSN to the last LSN for the index (the LSN for the flush log)
- * We mark the component flushing flag
- * We then move the write pointer to the next component and sets its first LSN to the flush log LSN
- */
- synchronized (this) {
- flushRequested[writeIndex] = true;
- writeIndex = (writeIndex + 1) % mutableLastLSNs.length;
- // Set the firstLSN of the next component unless it is being flushed
- if (writeIndex != readIndex) {
- firstLSNs[writeIndex] = mutableLastLSNs[writeIndex];
- }
-
- }
- }
- }
-
- @Override
- public void afterOperation(ILSMIndexOperationContext opCtx) throws HyracksDataException {
- //TODO: Copying Filters and all content of the metadata pages for flush operation should be done here
- if (opCtx.getNewComponent() == null) {
- // failed operation. Nothing to do.
- return;
- }
- putLSNIntoMetadata(opCtx.getNewComponent(), opCtx.getComponentsToBeMerged());
- putComponentIdIntoMetadata(opCtx.getIoOperationType(), opCtx.getNewComponent(),
- opCtx.getComponentsToBeMerged());
- componentLsnMap.put(opCtx.getNewComponent().getId(), getComponentLSN(opCtx.getComponentsToBeMerged()));
- if (opCtx.getIoOperationType() == LSMIOOperationType.MERGE) {
- if (opCtx.getComponentsToBeMerged().isEmpty()) {
- throw new IllegalStateException("Merge must have old components");
- }
- LongPointable markerLsn = LongPointable.FACTORY
- .createPointable(ComponentUtils.getLong(opCtx.getComponentsToBeMerged().get(0).getMetadata(),
- ComponentUtils.MARKER_LSN_KEY, ComponentUtils.NOT_FOUND, buffer));
- opCtx.getNewComponent().getMetadata().put(ComponentUtils.MARKER_LSN_KEY, markerLsn);
- } else if (opCtx.getIoOperationType() == LSMIOOperationType.FLUSH) {
- // advance memory component indexes
- synchronized (this) {
- // we've already consumed the specified LSN/component id.
- // Now we can advance to the next component
- flushRequested[readIndex] = false;
- // if the component which just finished flushing is the component that will be modified next,
- // we set its first LSN to its previous LSN
- if (readIndex == writeIndex) {
- firstLSNs[writeIndex] = mutableLastLSNs[writeIndex];
- }
- readIndex = (readIndex + 1) % mutableLastLSNs.length;
- }
- }
- }
-
- @Override
- public void afterFinalize(ILSMIndexOperationContext opCtx) throws HyracksDataException {
- // The operation was complete and the next I/O operation for the LSM index didn't start yet
- if (opCtx.getIoOperationType() == LSMIOOperationType.FLUSH) {
- hasFlushed = true;
- if (opCtx.getNewComponent() != null) {
- final Long lsn = componentLsnMap.remove(opCtx.getNewComponent().getId());
- if (lsn == null) {
- throw new IllegalStateException("Unidentified flushed component: " + opCtx.getNewComponent());
- }
- // empty component doesn't have any files
- final Optional<String> componentFile =
- opCtx.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);
- }
- }
- }
- }
-
- private void putLSNIntoMetadata(ILSMDiskComponent newComponent, List<? extends ILSMComponent> oldComponents)
- throws HyracksDataException {
- newComponent.getMetadata().put(LSN_KEY, LongPointable.FACTORY.createPointable(getComponentLSN(oldComponents)));
- }
-
- public static long getTreeIndexLSN(DiskComponentMetadata md) throws HyracksDataException {
- LongPointable pointable = new LongPointable();
- IMetadataPageManager metadataPageManager = md.getMetadataPageManager();
- metadataPageManager.get(metadataPageManager.createMetadataFrame(), LSN_KEY, pointable);
- return pointable.getLength() == 0 ? INVALID : pointable.longValue();
- }
-
- private ILSMComponentId getMergedComponentId(List<? extends ILSMComponent> mergedComponents)
- throws HyracksDataException {
- if (mergedComponents.isEmpty()) {
- return null;
- }
- return LSMComponentIdUtils.union(mergedComponents.get(0).getId(),
- mergedComponents.get(mergedComponents.size() - 1).getId());
-
- }
-
- private void putComponentIdIntoMetadata(LSMIOOperationType opType, ILSMDiskComponent newComponent,
- List<? extends ILSMComponent> oldComponents) throws HyracksDataException {
- // the id of flushed component is set when we copy the metadata of the memory component
- if (opType == LSMIOOperationType.MERGE) {
- ILSMComponentId componentId = getMergedComponentId(oldComponents);
- LSMComponentIdUtils.persist(componentId, newComponent.getMetadata());
- }
- }
-
- public synchronized void updateLastLSN(long lastLSN) {
- if (!flushRequested[writeIndex]) {
- //if the memory component pointed by writeIndex is being flushed, we should ignore this update call
- //since otherwise the original LSN is overwritten.
- //Moreover, since the memory component is already being flushed, the next scheduleFlush request must fail.
- //See https://issues.apache.org/jira/browse/ASTERIXDB-1917
- mutableLastLSNs[writeIndex] = lastLSN;
- if (hasFlushed || lsmIndex.isMemoryComponentsAllocated()) {
- // we only (re)set next component id if either this index has been flushed (no matter succeed or not)
- // or the memory component has been allocated
- // This prevents the case where indexes in a partition are being allocated, while another partition
- // tries to schedule flush
- nextComponentIds[writeIndex] = idGenerator.getId();
- }
- }
- }
-
- /**
- * Used during the recovery process to force refresh the next component id
- */
- public void forceRefreshNextId() {
- nextComponentIds[writeIndex] = idGenerator.getId();
- }
-
- public synchronized void setFirstLSN(long firstLSN) {
- // We make sure that this method is only called on an empty component so the first LSN is not set incorrectly
- firstLSNs[writeIndex] = firstLSN;
- }
-
- public synchronized long getFirstLSN() {
- // We make sure that this method is only called on a non-empty component so the returned LSN is meaningful
- // The firstLSN is always the lsn of the currently being flushed component or the next
- // to be flushed when no flush operation is on going
- return firstLSNs[readIndex];
- }
-
- public synchronized boolean hasPendingFlush() {
- for (int i = 0; i < flushRequested.length; i++) {
- if (flushRequested[i]) {
- return true;
- }
- }
- return false;
- }
-
- public long getComponentLSN(List<? extends ILSMComponent> diskComponents) throws HyracksDataException {
- if (diskComponents.isEmpty()) {
- // Implies a flush IO operation. --> moves the flush pointer
- // Flush operation of an LSM index are executed sequentially.
- synchronized (this) {
- return mutableLastLSNs[readIndex];
- }
- }
- // Get max LSN from the diskComponents. Implies a merge IO operation or Recovery operation.
- long maxLSN = -1L;
- for (ILSMComponent c : diskComponents) {
- DiskComponentMetadata md = ((ILSMDiskComponent) c).getMetadata();
- maxLSN = Math.max(getTreeIndexLSN(md), maxLSN);
- }
- return maxLSN;
- }
-
- private synchronized ILSMComponentId getLSMComponentId() {
- return nextComponentIds[recycleIndex];
- }
-
- @Override
- public void recycled(ILSMMemoryComponent component, boolean componentSwitched) throws HyracksDataException {
- ILSMComponentId componentId = getLSMComponentId();
- component.resetId(componentId, false);
- if (componentSwitched) {
- recycleIndex = (recycleIndex + 1) % nextComponentIds.length;
- }
- }
-
- @Override
- public void allocated(ILSMMemoryComponent component) throws HyracksDataException {
- if (component == lsmIndex.getCurrentMemoryComponent()) {
- // only set the component id for the first (current) memory component
- ILSMComponentId componentId = getLSMComponentId();
- component.resetId(componentId, false);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIndexIOOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIndexIOOperationCallbackFactory.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIndexIOOperationCallbackFactory.java
deleted file mode 100644
index 5b9883c..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIndexIOOperationCallbackFactory.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.asterix.common.ioopcallbacks;
-
-import java.io.ObjectStreamException;
-
-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.storage.am.lsm.common.api.ILSMComponentId;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGeneratorFactory;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory;
-import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentId;
-import org.apache.hyracks.storage.common.IResource;
-
-public abstract class AbstractLSMIndexIOOperationCallbackFactory implements ILSMIOOperationCallbackFactory {
-
- private static final long serialVersionUID = 1L;
-
- protected ILSMComponentIdGeneratorFactory idGeneratorFactory;
-
- protected transient INCServiceContext ncCtx;
-
- protected transient IResource resource;
-
- public AbstractLSMIndexIOOperationCallbackFactory(ILSMComponentIdGeneratorFactory idGeneratorFactory) {
- this.idGeneratorFactory = idGeneratorFactory;
- }
-
- @Override
- public void initialize(INCServiceContext ncCtx, IResource resource) {
- this.ncCtx = ncCtx;
- this.resource = resource;
- }
-
- protected ILSMComponentIdGenerator getComponentIdGenerator() throws HyracksDataException {
- return idGeneratorFactory.getComponentIdGenerator(ncCtx, resource);
- }
-
- protected IIndexCheckpointManagerProvider getIndexCheckpointManagerProvider() {
- return ((INcApplicationContext) ncCtx.getApplicationContext()).getIndexCheckpointManagerProvider();
- }
-
- private void readObjectNoData() throws ObjectStreamException {
- idGeneratorFactory = new ILSMComponentIdGeneratorFactory() {
- private static final long serialVersionUID = 1L;
-
- @Override
- public ILSMComponentIdGenerator getComponentIdGenerator(INCServiceContext serviceCtx, IResource resource) {
- // used for backward compatibility
- // if idGeneratorFactory is not set for legacy lsm indexes, we return a default
- // component id generator which always generates the missing component id.
- return new ILSMComponentIdGenerator() {
- @Override
- public void refresh() {
- // No op
- }
-
- @Override
- public ILSMComponentId getId() {
- return LSMComponentId.MISSING_COMPONENT_ID;
- }
- };
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallback.java
deleted file mode 100644
index db6c609..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallback.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.asterix.common.ioopcallbacks;
-
-import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
-
-public class LSMBTreeIOOperationCallback extends AbstractLSMIOOperationCallback {
-
- public LSMBTreeIOOperationCallback(ILSMIndex index, ILSMComponentIdGenerator idGenerator,
- IIndexCheckpointManagerProvider checkpointManagerProvider) {
- super(index, idGenerator, checkpointManagerProvider);
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallbackFactory.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallbackFactory.java
deleted file mode 100644
index 97badb2..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallbackFactory.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.asterix.common.ioopcallbacks;
-
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGeneratorFactory;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
-
-public class LSMBTreeIOOperationCallbackFactory extends AbstractLSMIndexIOOperationCallbackFactory {
-
- private static final long serialVersionUID = 1L;
-
- public LSMBTreeIOOperationCallbackFactory(ILSMComponentIdGeneratorFactory idGeneratorFactory) {
- super(idGeneratorFactory);
- }
-
- @Override
- public ILSMIOOperationCallback createIoOpCallback(ILSMIndex index) throws HyracksDataException {
- return new LSMBTreeIOOperationCallback(index, getComponentIdGenerator(), getIndexCheckpointManagerProvider());
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallback.java
deleted file mode 100644
index da1446b..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallback.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.asterix.common.ioopcallbacks;
-
-import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
-
-public class LSMBTreeWithBuddyIOOperationCallback extends AbstractLSMIOOperationCallback {
-
- public LSMBTreeWithBuddyIOOperationCallback(ILSMIndex lsmIndex, ILSMComponentIdGenerator idGenerator,
- IIndexCheckpointManagerProvider checkpointManagerProvider) {
- super(lsmIndex, idGenerator, checkpointManagerProvider);
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallbackFactory.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallbackFactory.java
deleted file mode 100644
index 9b32345..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallbackFactory.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.asterix.common.ioopcallbacks;
-
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGeneratorFactory;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
-
-public class LSMBTreeWithBuddyIOOperationCallbackFactory extends AbstractLSMIndexIOOperationCallbackFactory {
-
- private static final long serialVersionUID = 1L;
-
- public LSMBTreeWithBuddyIOOperationCallbackFactory(ILSMComponentIdGeneratorFactory idGeneratorFactory) {
- super(idGeneratorFactory);
- }
-
- @Override
- public ILSMIOOperationCallback createIoOpCallback(ILSMIndex index) throws HyracksDataException {
- return new LSMBTreeWithBuddyIOOperationCallback(index, getComponentIdGenerator(),
- getIndexCheckpointManagerProvider());
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..50f5906
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.asterix.common.ioopcallbacks;
+
+import java.util.ArrayDeque;
+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;
+import org.apache.asterix.common.storage.ResourceReference;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+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;
+import org.apache.hyracks.storage.am.common.freepage.MutableArrayValueReference;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationStatus;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationType;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+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.DiskComponentMetadata;
+import org.apache.hyracks.storage.am.lsm.common.impls.FlushOperation;
+import org.apache.hyracks.storage.am.lsm.common.util.ComponentUtils;
+import org.apache.hyracks.storage.am.lsm.common.util.LSMComponentIdUtils;
+
+// A single LSMIOOperationCallback per LSM index used to perform actions around Flush and Merge operations
+public class LSMIOOperationCallback implements ILSMIOOperationCallback {
+ public static final String KEY_FLUSH_LOG_LSN = "FlushLogLsn";
+ public static final String KEY_NEXT_COMPONENT_ID = "NextComponentId";
+ private static final String KEY_FIRST_LSN = "FirstLsn";
+ private static final MutableArrayValueReference KEY_METADATA_FLUSH_LOG_LSN =
+ new MutableArrayValueReference(KEY_FLUSH_LOG_LSN.getBytes());
+ public static final long INVALID_LSN = -1L;
+ private final ArrayBackedValueStorage buffer = new ArrayBackedValueStorage(Long.BYTES);
+ private final IIndexCheckpointManagerProvider indexCheckpointManagerProvider;
+ protected final DatasetInfo dsInfo;
+ protected final ILSMIndex lsmIndex;
+ private long firstLsnForCurrentMemoryComponent = 0L;
+ private long persistenceLsn = 0L;
+ private int pendingFlushes = 0;
+ private Deque<ILSMComponentId> componentIds = new ArrayDeque<>();
+
+ public LSMIOOperationCallback(DatasetInfo dsInfo, ILSMIndex lsmIndex, ILSMComponentId nextComponentId,
+ IIndexCheckpointManagerProvider indexCheckpointManagerProvider) {
+ this.dsInfo = dsInfo;
+ this.lsmIndex = lsmIndex;
+ this.indexCheckpointManagerProvider = indexCheckpointManagerProvider;
+ componentIds.add(nextComponentId);
+ }
+
+ @Override
+ public void beforeOperation(ILSMIOOperation operation) throws HyracksDataException {
+ // No Op
+ }
+
+ @Override
+ public void afterOperation(ILSMIOOperation operation) throws HyracksDataException {
+ if (operation.getStatus() == LSMIOOperationStatus.FAILURE) {
+ return;
+ }
+ if (operation.getIOOpertionType() == LSMIOOperationType.FLUSH) {
+ Map<String, Object> map = operation.getAccessor().getOpContext().getParameters();
+ putLSNIntoMetadata(operation.getNewComponent(), (Long) map.get(KEY_FLUSH_LOG_LSN));
+ putComponentIdIntoMetadata(operation.getNewComponent(),
+ ((FlushOperation) operation).getFlushingComponent().getId());
+ } else if (operation.getIOOpertionType() == LSMIOOperationType.MERGE) {
+ List<ILSMDiskComponent> mergedComponents = operation.getAccessor().getOpContext().getComponentsToBeMerged();
+ putLSNIntoMetadata(operation.getNewComponent(), mergedComponents);
+ putComponentIdIntoMetadata(operation.getNewComponent(), mergedComponents);
+ LongPointable markerLsn =
+ LongPointable.FACTORY.createPointable(ComponentUtils.getLong(mergedComponents.get(0).getMetadata(),
+ ComponentUtils.MARKER_LSN_KEY, ComponentUtils.NOT_FOUND, buffer));
+ operation.getNewComponent().getMetadata().put(ComponentUtils.MARKER_LSN_KEY, markerLsn);
+ }
+ }
+
+ @Override
+ public void afterFinalize(ILSMIOOperation operation) throws HyracksDataException {
+ if (operation.getStatus() == LSMIOOperationStatus.FAILURE) {
+ return;
+ }
+ if (operation.getIOOpertionType() == LSMIOOperationType.FLUSH) {
+ 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);
+ }
+ }
+ }
+
+ private void putLSNIntoMetadata(ILSMDiskComponent newComponent, List<ILSMDiskComponent> oldComponents)
+ throws HyracksDataException {
+ putLSNIntoMetadata(newComponent, getComponentLSN(oldComponents));
+ }
+
+ private void putLSNIntoMetadata(ILSMDiskComponent newComponent, long lsn) throws HyracksDataException {
+ newComponent.getMetadata().put(KEY_METADATA_FLUSH_LOG_LSN, LongPointable.FACTORY.createPointable(lsn));
+ }
+
+ public static long getTreeIndexLSN(DiskComponentMetadata md) throws HyracksDataException {
+ LongPointable pointable = new LongPointable();
+ IMetadataPageManager metadataPageManager = md.getMetadataPageManager();
+ metadataPageManager.get(metadataPageManager.createMetadataFrame(), KEY_METADATA_FLUSH_LOG_LSN, pointable);
+ return pointable.getLength() == 0 ? INVALID_LSN : pointable.longValue();
+ }
+
+ private ILSMComponentId getMergedComponentId(List<? extends ILSMComponent> mergedComponents)
+ throws HyracksDataException {
+ if (mergedComponents.isEmpty()) {
+ return null;
+ }
+ return LSMComponentIdUtils.union(mergedComponents.get(0).getId(),
+ mergedComponents.get(mergedComponents.size() - 1).getId());
+ }
+
+ private void putComponentIdIntoMetadata(ILSMDiskComponent newComponent, List<ILSMDiskComponent> oldComponents)
+ throws HyracksDataException {
+ ILSMComponentId componentId = getMergedComponentId(oldComponents);
+ putComponentIdIntoMetadata(newComponent, componentId);
+ }
+
+ private void putComponentIdIntoMetadata(ILSMDiskComponent newComponent, ILSMComponentId componentId)
+ throws HyracksDataException {
+ LSMComponentIdUtils.persist(componentId, newComponent.getMetadata());
+ }
+
+ /**
+ * Used during the recovery process to force refresh the next component id
+ */
+ public void forceRefreshNextId(ILSMComponentId nextComponentId) {
+ componentIds.clear();
+ componentIds.add(nextComponentId);
+ }
+
+ public synchronized void setFirstLsnForCurrentMemoryComponent(long firstLsn) {
+ this.firstLsnForCurrentMemoryComponent = firstLsn;
+ if (pendingFlushes == 0) {
+ this.persistenceLsn = firstLsn;
+ }
+ }
+
+ public synchronized long getPersistenceLsn() {
+ return persistenceLsn;
+ }
+
+ public long getComponentLSN(List<ILSMDiskComponent> diskComponents) throws HyracksDataException {
+ if (diskComponents.isEmpty()) {
+ throw new IllegalArgumentException("Can't get LSN from an empty list of disk components");
+ }
+ // Get max LSN from the diskComponents. Implies a merge IO operation or Recovery operation.
+ long maxLSN = -1L;
+ for (ILSMDiskComponent c : diskComponents) {
+ DiskComponentMetadata md = c.getMetadata();
+ maxLSN = Math.max(getTreeIndexLSN(md), maxLSN);
+ }
+ return maxLSN;
+ }
+
+ @Override
+ public void recycled(ILSMMemoryComponent component) throws HyracksDataException {
+ component.resetId(componentIds.poll(), false);
+ }
+
+ @Override
+ public synchronized void scheduled(ILSMIOOperation operation) throws HyracksDataException {
+ dsInfo.declareActiveIOOperation();
+ if (operation.getIOOpertionType() == LSMIOOperationType.FLUSH) {
+ pendingFlushes++;
+ Map<String, Object> map = operation.getAccessor().getOpContext().getParameters();
+ Long flushLsn = (Long) map.get(KEY_FLUSH_LOG_LSN);
+ map.put(KEY_FIRST_LSN, firstLsnForCurrentMemoryComponent);
+ componentIds.add((ILSMComponentId) map.get(KEY_NEXT_COMPONENT_ID));
+ firstLsnForCurrentMemoryComponent = flushLsn; // Advance the first lsn for new component
+ }
+ }
+
+ @Override
+ public synchronized void completed(ILSMIOOperation operation) {
+ if (operation.getIOOpertionType() == LSMIOOperationType.FLUSH) {
+ pendingFlushes--;
+ if (operation.getStatus() == LSMIOOperationStatus.SUCCESS) {
+ Map<String, Object> map = operation.getAccessor().getOpContext().getParameters();
+ persistenceLsn =
+ pendingFlushes == 0 ? firstLsnForCurrentMemoryComponent : (Long) map.get(KEY_FLUSH_LOG_LSN);
+ }
+ }
+ dsInfo.undeclareActiveIOOperation();
+ }
+
+ public synchronized boolean hasPendingFlush() {
+ return pendingFlushes > 0;
+ }
+
+ @Override
+ public void allocated(ILSMMemoryComponent component) throws HyracksDataException {
+ // No Op
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIndexIOOperationCallbackFactory.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..826ca50
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIndexIOOperationCallbackFactory.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.asterix.common.ioopcallbacks;
+
+import org.apache.asterix.common.api.IDatasetInfoProvider;
+import org.apache.asterix.common.api.ILSMComponentIdGeneratorFactory;
+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.storage.am.lsm.common.api.ILSMComponentIdGenerator;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.common.IResource;
+
+public class LSMIndexIOOperationCallbackFactory implements ILSMIOOperationCallbackFactory {
+
+ private static final long serialVersionUID = 1L;
+
+ private final ILSMComponentIdGeneratorFactory idGeneratorFactory;
+
+ protected final IDatasetInfoProvider datasetInfoProvider;
+
+ protected transient INCServiceContext ncCtx;
+
+ protected transient IResource resource;
+
+ public LSMIndexIOOperationCallbackFactory(ILSMComponentIdGeneratorFactory idGeneratorFactory,
+ IDatasetInfoProvider datasetInfoProvider) {
+ this.idGeneratorFactory = idGeneratorFactory;
+ this.datasetInfoProvider = datasetInfoProvider;
+ }
+
+ @Override
+ public void initialize(INCServiceContext ncCtx, IResource resource) {
+ this.ncCtx = ncCtx;
+ this.resource = resource;
+ }
+
+ protected ILSMComponentIdGenerator getComponentIdGenerator() throws HyracksDataException {
+ return idGeneratorFactory.getComponentIdGenerator(ncCtx, resource);
+ }
+
+ protected IIndexCheckpointManagerProvider getIndexCheckpointManagerProvider() {
+ return ((INcApplicationContext) ncCtx.getApplicationContext()).getIndexCheckpointManagerProvider();
+ }
+
+ @Override
+ public ILSMIOOperationCallback createIoOpCallback(ILSMIndex index) throws HyracksDataException {
+ return new LSMIOOperationCallback(datasetInfoProvider.getDatasetInfo(ncCtx), index,
+ getComponentIdGenerator().getId(), getIndexCheckpointManagerProvider());
+ }
+
+ @Override
+ public int getCurrentMemoryComponentIndex() throws HyracksDataException {
+ return idGeneratorFactory.getComponentIdGenerator(ncCtx, resource).getCurrentComponentIndex();
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallback.java
deleted file mode 100644
index 3ba9bcd..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallback.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.asterix.common.ioopcallbacks;
-
-import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
-
-public class LSMInvertedIndexIOOperationCallback extends AbstractLSMIOOperationCallback {
-
- public LSMInvertedIndexIOOperationCallback(ILSMIndex index, ILSMComponentIdGenerator idGenerator,
- IIndexCheckpointManagerProvider checkpointManagerProvider) {
- super(index, idGenerator, checkpointManagerProvider);
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallbackFactory.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallbackFactory.java
deleted file mode 100644
index 766ef95..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallbackFactory.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.asterix.common.ioopcallbacks;
-
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGeneratorFactory;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
-
-public class LSMInvertedIndexIOOperationCallbackFactory extends AbstractLSMIndexIOOperationCallbackFactory {
-
- private static final long serialVersionUID = 1L;
-
- public LSMInvertedIndexIOOperationCallbackFactory(ILSMComponentIdGeneratorFactory idGeneratorFactory) {
- super(idGeneratorFactory);
- }
-
- @Override
- public ILSMIOOperationCallback createIoOpCallback(ILSMIndex index) throws HyracksDataException {
- return new LSMInvertedIndexIOOperationCallback(index, getComponentIdGenerator(),
- getIndexCheckpointManagerProvider());
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallback.java
deleted file mode 100644
index f3e80ec..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallback.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.asterix.common.ioopcallbacks;
-
-import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
-
-public class LSMRTreeIOOperationCallback extends AbstractLSMIOOperationCallback {
-
- public LSMRTreeIOOperationCallback(ILSMIndex index, ILSMComponentIdGenerator idGenerator,
- IIndexCheckpointManagerProvider checkpointManagerProvodier) {
- super(index, idGenerator, checkpointManagerProvodier);
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallbackFactory.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallbackFactory.java
deleted file mode 100644
index 3a0afa8..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallbackFactory.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.asterix.common.ioopcallbacks;
-
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGeneratorFactory;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
-
-public class LSMRTreeIOOperationCallbackFactory extends AbstractLSMIndexIOOperationCallbackFactory {
-
- private static final long serialVersionUID = 1L;
-
- public LSMRTreeIOOperationCallbackFactory(ILSMComponentIdGeneratorFactory idGeneratorFactory) {
- super(idGeneratorFactory);
- }
-
- @Override
- public ILSMIOOperationCallback createIoOpCallback(ILSMIndex index) throws HyracksDataException {
- return new LSMRTreeIOOperationCallback(index, getComponentIdGenerator(), getIndexCheckpointManagerProvider());
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StorageConstants.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StorageConstants.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StorageConstants.java
index f6aa2c3..4bde9cf 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StorageConstants.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StorageConstants.java
@@ -41,7 +41,7 @@ public class StorageConstants {
/**
* The storage version of AsterixDB related artifacts (e.g. log files, checkpoint files, etc..).
*/
- private static final int LOCAL_STORAGE_VERSION = 2;
+ private static final int LOCAL_STORAGE_VERSION = 3;
/**
* The storage version of AsterixDB stack.
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/context/CorrelatedPrefixMergePolicyTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/context/CorrelatedPrefixMergePolicyTest.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/context/CorrelatedPrefixMergePolicyTest.java
index befbeed..843cb21 100644
--- a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/context/CorrelatedPrefixMergePolicyTest.java
+++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/context/CorrelatedPrefixMergePolicyTest.java
@@ -34,7 +34,6 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent.ComponentState;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
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.ILSMMergePolicy;
@@ -220,7 +219,7 @@ public class CorrelatedPrefixMergePolicyTest extends TestCase {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
- List<ILSMDiskComponent> mergedComponents = invocation.getArgumentAt(1, List.class);
+ List<ILSMDiskComponent> mergedComponents = invocation.getArgumentAt(0, List.class);
mergedComponents.forEach(component -> {
try {
resultComponentIDs.add(component.getId());
@@ -230,8 +229,7 @@ public class CorrelatedPrefixMergePolicyTest extends TestCase {
});
return null;
}
- }).when(accessor).scheduleMerge(Mockito.any(ILSMIOOperationCallback.class),
- Mockito.anyListOf(ILSMDiskComponent.class));
+ }).when(accessor).scheduleMerge(Mockito.anyListOf(ILSMDiskComponent.class));
Mockito.when(index.createAccessor(Mockito.any(IIndexAccessParameters.class))).thenReturn(accessor);
Mockito.when(index.isPrimaryIndex()).thenReturn(isPrimary);