merge master into branch.
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AbstractAqlTranslator.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AbstractAqlTranslator.java
index 9fe5750..3ca94c8 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AbstractAqlTranslator.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AbstractAqlTranslator.java
@@ -24,6 +24,7 @@
import edu.uci.ics.asterix.aql.expression.DropStatement;
import edu.uci.ics.asterix.aql.expression.InsertStatement;
import edu.uci.ics.asterix.aql.expression.NodeGroupDropStatement;
+import edu.uci.ics.asterix.common.config.AsterixClusterProperties;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
import edu.uci.ics.asterix.metadata.bootstrap.MetadataConstants;
import edu.uci.ics.asterix.metadata.dataset.hints.DatasetHints;
@@ -41,6 +42,12 @@
protected static final Map<String, BuiltinType> builtinTypeMap = AsterixBuiltinTypeMap.getBuiltinTypes();
public void validateOperation(Dataverse defaultDataverse, Statement stmt) throws AsterixException {
+
+ if (AsterixClusterProperties.INSTANCE.getState().equals(AsterixClusterProperties.State.UNUSABLE)) {
+ throw new AsterixException(" Asterix Cluster is in " + AsterixClusterProperties.State.UNUSABLE + " state."
+ + "\n One or more Node Controllers have left.\n");
+ }
+
boolean invalidOperation = false;
String message = null;
String dataverse = defaultDataverse != null ? defaultDataverse.getDataverseName() : null;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/AsterixAppRuntimeContext.java b/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/AsterixAppRuntimeContext.java
index 65eae64..e64f68f 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/AsterixAppRuntimeContext.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/AsterixAppRuntimeContext.java
@@ -49,10 +49,11 @@
import edu.uci.ics.hyracks.storage.common.file.ResourceIdFactory;
import edu.uci.ics.hyracks.storage.common.file.ResourceIdFactoryProvider;
-public class AsterixAppRuntimeContext implements IAsterixAppRuntimeContext, IAsterixPropertiesProvider {
+public class AsterixAppRuntimeContext implements IAsterixAppRuntimeContext, IAsterixPropertiesProvider {
private static final int DEFAULT_BUFFER_CACHE_PAGE_SIZE = 32768;
private static final int DEFAULT_LIFECYCLEMANAGER_MEMORY_BUDGET = 1024 * 1024 * 1024; // 1GB
private static final int DEFAULT_MAX_OPEN_FILES = Integer.MAX_VALUE;
+ private static final int METADATA_IO_DEVICE_ID = 0;
private final INCApplicationContext ncApplicationContext;
@@ -188,6 +189,10 @@
return ioManager;
}
+ public int getMetaDataIODeviceId() {
+ return METADATA_IO_DEVICE_ID;
+ }
+
@Override
public AsterixStorageProperties getStorageProperties() {
return storageProperties;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java b/asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java
index 911f6fd..99561e3 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java
@@ -61,6 +61,7 @@
import edu.uci.ics.asterix.aql.expression.WriteStatement;
import edu.uci.ics.asterix.aql.util.FunctionUtils;
import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
+import edu.uci.ics.asterix.common.config.AsterixClusterProperties;
import edu.uci.ics.asterix.common.config.GlobalConfig;
import edu.uci.ics.asterix.common.exceptions.ACIDException;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java
index 777ca6f..e22c215 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java
@@ -171,7 +171,7 @@
//prepare a LocalResourceMetadata which will be stored in NC's local resource repository
ILocalResourceMetadata localResourceMetadata = new LSMBTreeLocalResourceMetadata(typeTraits,
comparatorFactories, blooFilterKeyFields, true, storageProperties.getMemoryComponentPageSize(),
- storageProperties.getMemoryComponentNumPages());
+ storageProperties.getMemoryComponentNumPages(), fs);
ILocalResourceFactoryProvider localResourceFactoryProvider = new PersistentLocalResourceFactoryProvider(
localResourceMetadata, LocalResource.LSMBTreeResource);
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryBTreeCreator.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryBTreeCreator.java
index fb83871..ebd1a9c 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryBTreeCreator.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryBTreeCreator.java
@@ -40,7 +40,8 @@
//prepare a LocalResourceMetadata which will be stored in NC's local resource repository
ILocalResourceMetadata localResourceMetadata = new LSMBTreeLocalResourceMetadata(
secondaryRecDesc.getTypeTraits(), secondaryComparatorFactories, secondaryBloomFilterKeyFields, false,
- storageProperties.getMemoryComponentPageSize(), storageProperties.getMemoryComponentNumPages());
+ storageProperties.getMemoryComponentPageSize(), storageProperties.getMemoryComponentNumPages(),
+ secondaryFileSplitProvider.getFileSplits());
ILocalResourceFactoryProvider localResourceFactoryProvider = new PersistentLocalResourceFactoryProvider(
localResourceMetadata, LocalResource.LSMBTreeResource);
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryIndexCreator.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryIndexCreator.java
index 677eb23..b79102b 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryIndexCreator.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryIndexCreator.java
@@ -76,7 +76,8 @@
import edu.uci.ics.hyracks.storage.am.lsm.btree.dataflow.LSMBTreeDataflowHelperFactory;
@SuppressWarnings("rawtypes")
-// TODO: We should eventually have a hierarchy of classes that can create all possible index job specs,
+// TODO: We should eventually have a hierarchy of classes that can create all
+// possible index job specs,
// not just for creation.
public abstract class SecondaryIndexCreator {
protected final PhysicalOptimizationConfig physOptConf;
@@ -336,16 +337,13 @@
for (int i = 0; i < numSecondaryKeyFields + numPrimaryKeys; i++) {
fieldPermutation[i] = i;
}
- Pair<IFileSplitProvider, AlgebricksPartitionConstraint> secondarySplitsAndConstraint = metadataProvider
- .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(dataverseName, datasetName,
- secondaryIndexName);
TreeIndexBulkLoadOperatorDescriptor treeIndexBulkLoadOp = new TreeIndexBulkLoadOperatorDescriptor(spec,
AsterixRuntimeComponentsProvider.LSMBTREE_PROVIDER, AsterixRuntimeComponentsProvider.LSMBTREE_PROVIDER,
- secondarySplitsAndConstraint.first, secondaryRecDesc.getTypeTraits(), secondaryComparatorFactories,
+ secondaryFileSplitProvider, secondaryRecDesc.getTypeTraits(), secondaryComparatorFactories,
secondaryBloomFilterKeyFields, fieldPermutation, fillFactor, false, numElementsHint,
dataflowHelperFactory, NoOpOperationCallbackFactory.INSTANCE);
AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, treeIndexBulkLoadOp,
- secondarySplitsAndConstraint.second);
+ secondaryPartitionConstraint);
return treeIndexBulkLoadOp;
}
@@ -365,7 +363,8 @@
}
ICopyEvaluatorFactory selectCond = null;
if (numSecondaryKeyFields > 1) {
- // Create conjunctive condition where all secondary index keys must satisfy 'is not null'.
+ // Create conjunctive condition where all secondary index keys must
+ // satisfy 'is not null'.
AndDescriptor andDesc = new AndDescriptor();
selectCond = andDesc.createEvaluatorFactory(andArgsEvalFactories);
} else {
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryInvertedIndexCreator.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryInvertedIndexCreator.java
index a2e28da..ef94ec2 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryInvertedIndexCreator.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryInvertedIndexCreator.java
@@ -157,7 +157,7 @@
ILocalResourceMetadata localResourceMetadata = new LSMInvertedIndexLocalResourceMetadata(invListsTypeTraits,
primaryComparatorFactories, tokenTypeTraits, tokenComparatorFactories, tokenizerFactory,
storageProperties.getMemoryComponentPageSize(), storageProperties.getMemoryComponentNumPages(),
- isPartitioned);
+ isPartitioned, secondaryFileSplitProvider.getFileSplits());
ILocalResourceFactoryProvider localResourceFactoryProvider = new PersistentLocalResourceFactoryProvider(
localResourceMetadata, LocalResource.LSMInvertedIndexResource);
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryRTreeCreator.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryRTreeCreator.java
index ad61cbe..b7e4886 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryRTreeCreator.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryRTreeCreator.java
@@ -65,7 +65,7 @@
secondaryRecDesc.getTypeTraits(), secondaryComparatorFactories, primaryComparatorFactories,
valueProviderFactories, RTreePolicyType.RTREE, AqlMetadataProvider.proposeLinearizer(keyType,
secondaryComparatorFactories.length), storageProperties.getMemoryComponentPageSize(),
- storageProperties.getMemoryComponentNumPages());
+ storageProperties.getMemoryComponentNumPages(), secondaryFileSplitProvider.getFileSplits());
ILocalResourceFactoryProvider localResourceFactoryProvider = new PersistentLocalResourceFactoryProvider(
localResourceMetadata, LocalResource.LSMRTreeResource);
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/CCApplicationEntryPoint.java b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/CCApplicationEntryPoint.java
index 8019a41..5d4f6ec 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/CCApplicationEntryPoint.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/CCApplicationEntryPoint.java
@@ -54,6 +54,7 @@
webServer.start();
setupJSONAPIServer(externalProperties);
jsonAPIServer.start();
+ ccAppCtx.addClusterLifecycleListener(ClusterLifecycleListener.INSTANCE);
}
@Override
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/ClusterLifecycleListener.java b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/ClusterLifecycleListener.java
new file mode 100644
index 0000000..d1ea1dc
--- /dev/null
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/ClusterLifecycleListener.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ *
+ * 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 edu.uci.ics.asterix.hyracks.bootstrap;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import edu.uci.ics.asterix.common.config.AsterixClusterProperties;
+import edu.uci.ics.hyracks.api.application.IClusterLifecycleListener;
+
+public class ClusterLifecycleListener implements IClusterLifecycleListener {
+
+ public static ClusterLifecycleListener INSTANCE = new ClusterLifecycleListener();
+
+ private ClusterLifecycleListener() {
+ }
+
+ private static final Logger LOGGER = Logger.getLogger(ClusterLifecycleListener.class.getName());
+
+ @Override
+ public void notifyNodeJoin(String nodeId, Map<String, String> ncConfiguration) {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("NC: " + nodeId + " joined");
+ }
+ AsterixClusterProperties.INSTANCE.addNCConfiguration(nodeId, ncConfiguration);
+ }
+
+ public void notifyNodeFailure(Set<String> deadNodeIds) {
+ for (String deadNode : deadNodeIds) {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("NC: " + deadNode + " left");
+ }
+ AsterixClusterProperties.INSTANCE.removeNCConfiguration(deadNode);
+ }
+
+ }
+
+}
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/config/AsterixClusterProperties.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/config/AsterixClusterProperties.java
new file mode 100644
index 0000000..362774f
--- /dev/null
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/config/AsterixClusterProperties.java
@@ -0,0 +1,84 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ *
+ * 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 edu.uci.ics.asterix.common.config;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import edu.uci.ics.asterix.common.api.AsterixAppContextInfo;
+
+public class AsterixClusterProperties {
+
+ private static final Logger LOGGER = Logger.getLogger(AsterixClusterProperties.class.getName());
+
+ private static final String IO_DEVICES = "iodevices";
+
+ public static final AsterixClusterProperties INSTANCE = new AsterixClusterProperties();
+
+ private Map<String, Map<String, String>> ncConfiguration = new HashMap<String, Map<String, String>>();
+
+ private AsterixClusterProperties() {
+ }
+
+ public enum State {
+ ACTIVE,
+ UNUSABLE
+ }
+
+ private State state = State.UNUSABLE;
+
+ public void removeNCConfiguration(String nodeId) {
+ state = State.UNUSABLE;
+ ncConfiguration.remove(nodeId);
+ }
+
+ public void addNCConfiguration(String nodeId, Map<String, String> configuration) {
+ ncConfiguration.put(nodeId, configuration);
+ if (ncConfiguration.keySet().size() == AsterixAppContextInfo.getInstance().getMetadataProperties()
+ .getNodeNames().size()) {
+ state = State.ACTIVE;
+ }
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info(" Registering configuration parameters for node id" + nodeId);
+ }
+ }
+
+ /**
+ * Returns the number of IO devices configured for a Node Controller
+ *
+ * @param nodeId
+ * unique identifier of the Node Controller
+ * @return number of IO devices. -1 if the node id is not valid. A node id is not valid
+ * if it does not correspond to the set of registered Node Controllers.
+ */
+ public int getNumberOfIODevices(String nodeId) {
+ Map<String, String> ncConfig = ncConfiguration.get(nodeId);
+ if (ncConfig == null) {
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Configuration parameters for nodeId" + nodeId
+ + " not found. The node has not joined yet or has left.");
+ }
+ return -1;
+ }
+ return ncConfig.get(IO_DEVICES).split(",").length;
+ }
+
+ public State getState() {
+ return state;
+ }
+
+}
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java
index ee05c7a..d0f6040 100644
--- a/asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java
@@ -22,14 +22,13 @@
* Provides methods for obtaining the IIndexRegistryProvider, IStorageManager and
* ICCApplicationContext implementation.
*/
-public interface IAsterixApplicationContextInfo{
+public interface IAsterixApplicationContextInfo {
/**
*
*/
public IIndexLifecycleManagerProvider getIndexLifecycleManagerProvider();
-
/**
* Returns an instance of the implementation for IStorageManagerInterface.
*
@@ -43,5 +42,4 @@
* @return ICCApplicationContext implementation instance
*/
public ICCApplicationContext getCCApplicationContext();
-
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java
index 791666d..41a9230b 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java
@@ -333,7 +333,8 @@
}
private static void enlistMetadataDataset(IMetadataIndex index, boolean create) throws Exception {
- String filePath = metadataStore + File.separator + index.getFileNameRelativePath();
+ String filePath = metadataStore + File.separator + index.getFileNameRelativePath() + File.separator
+ + "device_id_" + runtimeContext.getMetaDataIODeviceId();
FileReference file = new FileReference(new File(filePath));
IInMemoryBufferCache memBufferCache = new InMemoryBufferCache(new HeapBufferAllocator(), DEFAULT_MEM_PAGE_SIZE,
DEFAULT_MEM_NUM_PAGES, new TransientFileMapManager());
@@ -350,7 +351,7 @@
fileMapProvider, typeTraits, comparatorFactories, bloomFilterKeyFields,
runtimeContext.getBloomFilterFalsePositiveRate(), runtimeContext.getLSMMergePolicy(),
runtimeContext.getLSMBTreeOperationTrackerFactory(), runtimeContext.getLSMIOScheduler(),
- AsterixRuntimeComponentsProvider.LSMBTREE_PROVIDER);
+ AsterixRuntimeComponentsProvider.LSMBTREE_PROVIDER, runtimeContext.getMetaDataIODeviceId());
lsmBtree.create();
resourceID = runtimeContext.getResourceIdFactory().createId();
indexLifecycleManager.register(resourceID, lsmBtree);
@@ -358,12 +359,14 @@
AsterixStorageProperties storageProperties = propertiesProvider.getStorageProperties();
ILocalResourceMetadata localResourceMetadata = new LSMBTreeLocalResourceMetadata(typeTraits,
comparatorFactories, bloomFilterKeyFields, index.isPrimaryIndex(),
- storageProperties.getMemoryComponentPageSize(), storageProperties.getMemoryComponentNumPages());
+ storageProperties.getMemoryComponentPageSize(), storageProperties.getMemoryComponentNumPages(),
+ runtimeContext.getMetaDataIODeviceId());
ILocalResourceFactoryProvider localResourceFactoryProvider = new PersistentLocalResourceFactoryProvider(
localResourceMetadata, LocalResource.LSMBTreeResource);
ILocalResourceFactory localResourceFactory = localResourceFactoryProvider.getLocalResourceFactory();
- localResourceRepository.insert(localResourceFactory.createLocalResource(resourceID, file.getFile()
- .getPath(), 0));
+ localResourceRepository.insert(
+ localResourceFactory.createLocalResource(resourceID, file.getFile().getPath(), 0),
+ runtimeContext.getMetaDataIODeviceId());
} else {
resourceID = localResourceRepository.getResourceByName(file.getFile().getPath()).getResourceId();
lsmBtree = (LSMBTree) indexLifecycleManager.getIndex(resourceID);
@@ -372,7 +375,7 @@
bufferCache, fileMapProvider, typeTraits, comparatorFactories, bloomFilterKeyFields,
runtimeContext.getBloomFilterFalsePositiveRate(), runtimeContext.getLSMMergePolicy(),
runtimeContext.getLSMBTreeOperationTrackerFactory(), runtimeContext.getLSMIOScheduler(),
- AsterixRuntimeComponentsProvider.LSMBTREE_PROVIDER);
+ AsterixRuntimeComponentsProvider.LSMBTREE_PROVIDER, runtimeContext.getMetaDataIODeviceId());
indexLifecycleManager.register(resourceID, lsmBtree);
}
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java
index 3128153..57e9b0d 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java
@@ -23,6 +23,7 @@
import java.util.Map;
import java.util.logging.Logger;
+import edu.uci.ics.asterix.common.config.AsterixClusterProperties;
import edu.uci.ics.asterix.common.config.AsterixStorageProperties;
import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
import edu.uci.ics.asterix.common.config.DatasetConfig.IndexType;
@@ -418,7 +419,8 @@
adapterFactoryClassname = adapterFactoryMapping.get(adapterName);
if (adapterFactoryClassname != null) {
} else {
- // adapterName has been provided as a fully qualified classname
+ // adapterName has been provided as a fully qualified
+ // classname
adapterFactoryClassname = adapterName;
}
adapterFactory = (IAdapterFactory) Class.forName(adapterFactoryClassname).newInstance();
@@ -769,9 +771,11 @@
numElementsHint = Long.parseLong(dataset.getHints().get("CARDINALITY"));
}
- //TODO
- //figure out the right behavior of the bulkload and then give the right callback
- //(ex. what's the expected behavior when there is an error during bulkload?)
+ // TODO
+ // figure out the right behavior of the bulkload and then give the
+ // right callback
+ // (ex. what's the expected behavior when there is an error during
+ // bulkload?)
TreeIndexBulkLoadOperatorDescriptor btreeBulkLoad = new TreeIndexBulkLoadOperatorDescriptor(spec,
appContext.getStorageManagerInterface(), appContext.getIndexLifecycleManagerProvider(),
splitsAndConstraint.first, typeTraits, comparatorFactories, bloomFilterKeyFields, fieldPermutation,
@@ -830,7 +834,7 @@
Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint = splitProviderAndPartitionConstraintsForInternalOrFeedDataset(
dataSource.getId().getDataverseName(), datasetName, indexName);
- //prepare callback
+ // prepare callback
JobId jobId = ((JobEventListenerFactory) spec.getJobletEventListenerFactory()).getJobId();
int datasetId = dataset.getDatasetId();
int[] primaryKeyFields = new int[numKeys];
@@ -1023,7 +1027,7 @@
Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint = splitProviderAndPartitionConstraintsForInternalOrFeedDataset(
dataverseName, datasetName, indexName);
- //prepare callback
+ // prepare callback
JobId jobId = ((JobEventListenerFactory) spec.getJobletEventListenerFactory()).getJobId();
int datasetId = dataset.getDatasetId();
int[] primaryKeyFields = new int[primaryKeys.size()];
@@ -1138,7 +1142,8 @@
tokenComparatorFactories[0] = NonTaggedFormatUtil.getTokenBinaryComparatorFactory(secondaryKeyType);
tokenTypeTraits[0] = NonTaggedFormatUtil.getTokenTypeTrait(secondaryKeyType);
if (isPartitioned) {
- // The partitioning field is hardcoded to be a short *without* an Asterix type tag.
+ // The partitioning field is hardcoded to be a short *without*
+ // an Asterix type tag.
tokenComparatorFactories[1] = PointableBinaryComparatorFactory.of(ShortPointable.FACTORY);
tokenTypeTraits[1] = ShortPointable.TYPE_TRAITS;
}
@@ -1149,7 +1154,7 @@
Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint = splitProviderAndPartitionConstraintsForInternalOrFeedDataset(
dataverseName, datasetName, indexName);
- //prepare callback
+ // prepare callback
JobId jobId = ((JobEventListenerFactory) spec.getJobletEventListenerFactory()).getJobId();
int datasetId = dataset.getDatasetId();
int[] primaryKeyFields = new int[primaryKeys.size()];
@@ -1244,7 +1249,7 @@
Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint = splitProviderAndPartitionConstraintsForInternalOrFeedDataset(
dataverseName, datasetName, indexName);
- //prepare callback
+ // prepare callback
JobId jobId = ((JobEventListenerFactory) spec.getJobletEventListenerFactory()).getJobId();
int datasetId = dataset.getDatasetId();
int[] primaryKeyFields = new int[numPrimaryKeys];
@@ -1338,8 +1343,13 @@
continue;
}
for (int i = 0; i < nodeStores.length; i++) {
- File f = new File(nodeStores[i] + File.separator + relPathFile);
- splits.add(new FileSplit(node, new FileReference(f)));
+ int numIODevices = AsterixClusterProperties.INSTANCE.getNumberOfIODevices(node);
+ for (int j = 0; j < nodeStores.length; j++) {
+ for (int k = 0; k < numIODevices; k++) {
+ File f = new File(nodeStores[j] + File.separator + relPathFile);
+ splits.add(new FileSplit(node, new FileReference(f), k));
+ }
+ }
}
}
return splits.toArray(new FileSplit[] {});
@@ -1368,9 +1378,17 @@
LOGGER.warning("Node " + nd + " has no stores.");
throw new AlgebricksException("Node " + nd + " has no stores.");
} else {
+ int numIODevices;
+ if (datasetDetails.getNodeGroupName().compareTo(MetadataConstants.METADATA_NODEGROUP_NAME) == 0) {
+ numIODevices = 1;
+ } else {
+ numIODevices = AsterixClusterProperties.INSTANCE.getNumberOfIODevices(nd);
+ }
for (int j = 0; j < nodeStores.length; j++) {
- File f = new File(nodeStores[j] + File.separator + relPathFile);
- splitArray.add(new FileSplit(nd, new FileReference(f)));
+ for (int k = 0; k < numIODevices; k++) {
+ File f = new File(nodeStores[j] + File.separator + relPathFile);
+ splitArray.add(new FileSplit(nd, new FileReference(f), k));
+ }
}
}
}
@@ -1417,21 +1435,6 @@
return dataverseName + File.separator + fileName;
}
- public Pair<IFileSplitProvider, IFileSplitProvider> getInvertedIndexFileSplitProviders(
- IFileSplitProvider splitProvider) {
- int numSplits = splitProvider.getFileSplits().length;
- FileSplit[] btreeSplits = new FileSplit[numSplits];
- FileSplit[] invListsSplits = new FileSplit[numSplits];
- for (int i = 0; i < numSplits; i++) {
- String nodeName = splitProvider.getFileSplits()[i].getNodeName();
- String path = splitProvider.getFileSplits()[i].getLocalFile().getFile().getPath();
- btreeSplits[i] = new FileSplit(nodeName, path + "_$btree");
- invListsSplits[i] = new FileSplit(nodeName, path + "_$invlists");
- }
- return new Pair<IFileSplitProvider, IFileSplitProvider>(new ConstantFileSplitProvider(btreeSplits),
- new ConstantFileSplitProvider(invListsSplits));
- }
-
public Dataset findDataset(String dataverse, String dataset) throws AlgebricksException {
try {
return MetadataManager.INSTANCE.getDataset(mdTxnCtx, dataverse, dataset);
@@ -1463,10 +1466,13 @@
public AlgebricksPartitionConstraint getClusterLocations() {
ArrayList<String> locs = new ArrayList<String>();
- for (String k : stores.keySet()) {
- String[] nodeStores = stores.get(k);
+ for (String i : stores.keySet()) {
+ String[] nodeStores = stores.get(i);
+ int numIODevices = AsterixClusterProperties.INSTANCE.getNumberOfIODevices(i);
for (int j = 0; j < nodeStores.length; j++) {
- locs.add(k);
+ for (int k = 0; k < numIODevices; k++) {
+ locs.add(i);
+ }
}
}
String[] cluster = new String[locs.size()];
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixAppContextInfo.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixAppContextInfo.java
index e85a72d..81100bf 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixAppContextInfo.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixAppContextInfo.java
@@ -57,7 +57,6 @@
INSTANCE.metadataProperties = new AsterixMetadataProperties(propertiesAccessor);
INSTANCE.storageProperties = new AsterixStorageProperties(propertiesAccessor);
INSTANCE.txnProperties = new AsterixTransactionProperties(propertiesAccessor);
-
Logger.getLogger("edu.uci.ics").setLevel(INSTANCE.externalProperties.getLogLevel());
}
@@ -108,6 +107,4 @@
public IStorageManagerInterface getStorageManagerInterface() {
return AsterixRuntimeComponentsProvider.NOINDEX_PROVIDER;
}
-
-
}
diff --git a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadata.java b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadata.java
index e06b5df..c3f9747 100644
--- a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadata.java
+++ b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadata.java
@@ -6,6 +6,7 @@
import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
import edu.uci.ics.hyracks.storage.am.common.api.IInMemoryFreePageManager;
import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
@@ -27,14 +28,27 @@
private final int[] bloomFilterKeyFields;
private final int memPageSize;
private final int memNumPages;
+ private FileSplit[] fileSplits;
+ private int ioDeviceID;
public LSMBTreeLocalResourceMetadata(ITypeTraits[] typeTraits, IBinaryComparatorFactory[] cmpFactories,
- int[] bloomFilterKeyFields, boolean isPrimary, int memPageSize, int memNumPages) {
+ int[] bloomFilterKeyFields, boolean isPrimary, int memPageSize, int memNumPages, FileSplit[] fileSplits) {
this.typeTraits = typeTraits;
this.cmpFactories = cmpFactories;
this.bloomFilterKeyFields = bloomFilterKeyFields;
this.memPageSize = memPageSize;
this.memNumPages = memNumPages;
+ this.fileSplits = fileSplits;
+ }
+
+ public LSMBTreeLocalResourceMetadata(ITypeTraits[] typeTraits, IBinaryComparatorFactory[] cmpFactories,
+ int[] bloomFilterKeyFields, boolean isPrimary, int memPageSize, int memNumPages, int ioDeviceID) {
+ this.typeTraits = typeTraits;
+ this.cmpFactories = cmpFactories;
+ this.bloomFilterKeyFields = bloomFilterKeyFields;
+ this.memPageSize = memPageSize;
+ this.memNumPages = memNumPages;
+ this.ioDeviceID = ioDeviceID;
}
@Override
@@ -45,13 +59,13 @@
memNumPages, new TransientFileMapManager());
ITreeIndexMetaDataFrameFactory metaDataFrameFactory = new LIFOMetaDataFrameFactory();
IInMemoryFreePageManager memFreePageManager = new InMemoryFreePageManager(memNumPages, metaDataFrameFactory);
- LSMBTree lsmBTree = LSMBTreeUtils.createLSMTree(memBufferCache, memFreePageManager,
- runtimeContextProvider.getIOManager(), file, runtimeContextProvider.getBufferCache(),
- runtimeContextProvider.getFileMapManager(), typeTraits, cmpFactories, bloomFilterKeyFields,
- runtimeContextProvider.getBloomFilterFalsePositiveRate(), runtimeContextProvider.getLSMMergePolicy(),
- runtimeContextProvider.getLSMBTreeOperationTrackerFactory(),
- runtimeContextProvider.getLSMIOScheduler(),
- runtimeContextProvider.getLSMBTreeIOOperationCallbackProvider(), partition);
+ LSMBTree lsmBTree = LSMBTreeUtils.createLSMTree(memBufferCache, memFreePageManager, runtimeContextProvider
+ .getIOManager(), file, runtimeContextProvider.getBufferCache(), runtimeContextProvider
+ .getFileMapManager(), typeTraits, cmpFactories, bloomFilterKeyFields, runtimeContextProvider
+ .getBloomFilterFalsePositiveRate(), runtimeContextProvider.getLSMMergePolicy(), runtimeContextProvider
+ .getLSMBTreeOperationTrackerFactory(), runtimeContextProvider.getLSMIOScheduler(),
+ runtimeContextProvider.getLSMBTreeIOOperationCallbackProvider(), fileSplits == null ? ioDeviceID
+ : fileSplits[partition].getIODeviceId());
return lsmBTree;
}
diff --git a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadata.java b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadata.java
index d56c6ee..89a32c5 100644
--- a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadata.java
+++ b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadata.java
@@ -4,6 +4,7 @@
import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
import edu.uci.ics.hyracks.storage.am.common.api.IInMemoryFreePageManager;
import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
@@ -28,11 +29,12 @@
private final int memPageSize;
private final int memNumPages;
private final boolean isPartitioned;
+ private final FileSplit[] fileSplits;
public LSMInvertedIndexLocalResourceMetadata(ITypeTraits[] invListTypeTraits,
IBinaryComparatorFactory[] invListCmpFactories, ITypeTraits[] tokenTypeTraits,
IBinaryComparatorFactory[] tokenCmpFactories, IBinaryTokenizerFactory tokenizerFactory, int memPageSize,
- int memNumPages, boolean isPartitioned) {
+ int memNumPages, boolean isPartitioned, FileSplit[] fileSplits) {
this.invListTypeTraits = invListTypeTraits;
this.invListCmpFactories = invListCmpFactories;
this.tokenTypeTraits = tokenTypeTraits;
@@ -41,6 +43,7 @@
this.memPageSize = memPageSize;
this.memNumPages = memNumPages;
this.isPartitioned = isPartitioned;
+ this.fileSplits = fileSplits;
}
@Override
@@ -62,7 +65,8 @@
runtimeContextProvider.getLSMMergePolicy(),
runtimeContextProvider.getLSMInvertedIndexOperationTrackerFactory(),
runtimeContextProvider.getLSMIOScheduler(),
- runtimeContextProvider.getLSMInvertedIndexIOOperationCallbackProvider(), partition);
+ runtimeContextProvider.getLSMInvertedIndexIOOperationCallbackProvider(),
+ fileSplits[partition].getIODeviceId());
} else {
return InvertedIndexUtils.createLSMInvertedIndex(memBufferCache, memFreePageManager,
runtimeContextProvider.getFileMapManager(), invListTypeTraits, invListCmpFactories,
@@ -72,7 +76,8 @@
runtimeContextProvider.getLSMMergePolicy(),
runtimeContextProvider.getLSMInvertedIndexOperationTrackerFactory(),
runtimeContextProvider.getLSMIOScheduler(),
- runtimeContextProvider.getLSMInvertedIndexIOOperationCallbackProvider(), partition);
+ runtimeContextProvider.getLSMInvertedIndexIOOperationCallbackProvider(),
+ fileSplits[partition].getIODeviceId());
}
} catch (IndexException e) {
throw new HyracksDataException(e);
diff --git a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadata.java b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadata.java
index 95fab45..2506697 100644
--- a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadata.java
+++ b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadata.java
@@ -8,6 +8,7 @@
import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
import edu.uci.ics.hyracks.storage.am.common.api.IInMemoryFreePageManager;
import edu.uci.ics.hyracks.storage.am.common.api.IPrimitiveValueProviderFactory;
import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
@@ -33,11 +34,12 @@
private final ILinearizeComparatorFactory linearizeCmpFactory;
private final int memPageSize;
private final int memNumPages;
+ private final FileSplit[] fileSplits;
public LSMRTreeLocalResourceMetadata(ITypeTraits[] typeTraits, IBinaryComparatorFactory[] rtreeCmpFactories,
IBinaryComparatorFactory[] btreeCmpFactories, IPrimitiveValueProviderFactory[] valueProviderFactories,
RTreePolicyType rtreePolicyType, ILinearizeComparatorFactory linearizeCmpFactory, int memPageSize,
- int memNumPages) {
+ int memNumPages, FileSplit[] fileSplits) {
this.typeTraits = typeTraits;
this.rtreeCmpFactories = rtreeCmpFactories;
this.btreeCmpFactories = btreeCmpFactories;
@@ -46,6 +48,7 @@
this.linearizeCmpFactory = linearizeCmpFactory;
this.memPageSize = memPageSize;
this.memNumPages = memNumPages;
+ this.fileSplits = fileSplits;
}
@Override
@@ -66,7 +69,8 @@
runtimeContextProvider.getLSMMergePolicy(),
runtimeContextProvider.getLSMRTreeOperationTrackerFactory(),
runtimeContextProvider.getLSMIOScheduler(),
- runtimeContextProvider.getLSMRTreeIOOperationCallbackProvider(), linearizeCmpFactory, partition);
+ runtimeContextProvider.getLSMRTreeIOOperationCallbackProvider(), linearizeCmpFactory,
+ fileSplits[partition].getIODeviceId());
} catch (TreeIndexException e) {
throw new HyracksDataException(e);
}
diff --git a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/PersistentLocalResourceRepository.java b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
index 191bc8a..dc42d06 100644
--- a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
+++ b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
@@ -29,6 +29,7 @@
import java.util.logging.Logger;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.api.io.IODeviceHandle;
import edu.uci.ics.hyracks.storage.common.file.ILocalResourceRepository;
import edu.uci.ics.hyracks.storage.common.file.LocalResource;
import edu.uci.ics.hyracks.storage.common.file.ResourceIdFactory;
@@ -36,25 +37,29 @@
public class PersistentLocalResourceRepository implements ILocalResourceRepository {
private static final Logger LOGGER = Logger.getLogger(PersistentLocalResourceRepository.class.getName());
- private final String mountPoint;
+ private final String[] mountPoints;
private static final String ROOT_METADATA_DIRECTORY = "asterix_root_metadata/";
private static final String ROOT_METADATA_FILE_NAME_PREFIX = ".asterix_root_metadata_";
private static final long ROOT_LOCAL_RESOURCE_ID = -4321;
private static final String METADATA_FILE_NAME = ".metadata";
private Map<String, LocalResource> name2ResourceMap = new HashMap<String, LocalResource>();
private Map<Long, LocalResource> id2ResourceMap = new HashMap<Long, LocalResource>();
- private String rootMetadataFileName;
- private String rootDir;
+ private final int numIODevices;
- public PersistentLocalResourceRepository(String mountPoint) throws HyracksDataException {
- File mountPointDir = new File(mountPoint);
- if (!mountPointDir.exists()) {
- throw new HyracksDataException(mountPointDir.getAbsolutePath() + "doesn't exist.");
- }
- if (!mountPoint.endsWith(System.getProperty("file.separator"))) {
- this.mountPoint = new String(mountPoint + System.getProperty("file.separator"));
- } else {
- this.mountPoint = new String(mountPoint);
+ public PersistentLocalResourceRepository(List<IODeviceHandle> devices) throws HyracksDataException {
+ numIODevices = devices.size();
+ this.mountPoints = new String[numIODevices];
+ for (int i = 0; i < numIODevices; i++) {
+ String mountPoint = devices.get(i).getPath().getPath();
+ File mountPointDir = new File(mountPoint);
+ if (!mountPointDir.exists()) {
+ throw new HyracksDataException(mountPointDir.getAbsolutePath() + "doesn't exist.");
+ }
+ if (!mountPoint.endsWith(System.getProperty("file.separator"))) {
+ mountPoints[i] = new String(mountPoint + System.getProperty("file.separator"));
+ } else {
+ mountPoints[i] = new String(mountPoint);
+ }
}
}
@@ -63,53 +68,40 @@
if (LOGGER.isLoggable(Level.INFO)) {
LOGGER.info("Initializing local resource repository ... ");
}
- LocalResource rootLocalResource = null;
- //#. if the rootMetadataFile doesn't exist, create it and return.
- rootMetadataFileName = new String(mountPoint + ROOT_METADATA_DIRECTORY + ROOT_METADATA_FILE_NAME_PREFIX
- + nodeId);
- File rootMetadataFile = new File(rootMetadataFileName);
if (isNewUniverse) {
- File rootMetadataDir = new File(mountPoint + ROOT_METADATA_DIRECTORY);
- if (!rootMetadataDir.exists()) {
- rootMetadataDir.mkdir();
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("created the root-metadata-file's directory: " + rootMetadataDir.getAbsolutePath());
+ //#. if the rootMetadataFile doesn't exist, create it and return.
+ for (int i = 0; i < numIODevices; i++) {
+ String rootMetadataFileName = new String(mountPoints[i] + ROOT_METADATA_DIRECTORY
+ + ROOT_METADATA_FILE_NAME_PREFIX + nodeId);
+ File rootMetadataFile = new File(rootMetadataFileName);
+
+ File rootMetadataDir = new File(mountPoints[i] + ROOT_METADATA_DIRECTORY);
+ if (!rootMetadataDir.exists()) {
+ rootMetadataDir.mkdir();
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("created the root-metadata-file's directory: " + rootMetadataDir.getAbsolutePath());
+ }
}
- }
- rootMetadataFile.delete();
- if (rootDir.startsWith(System.getProperty("file.separator"))) {
- this.rootDir = new String(mountPoint + rootDir.substring(System.getProperty("file.separator").length()));
- } else {
- this.rootDir = new String(mountPoint + rootDir);
- }
- rootLocalResource = new LocalResource(ROOT_LOCAL_RESOURCE_ID, rootMetadataFileName, 0, 0, this.rootDir);
- insert(rootLocalResource);
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("created the root-metadata-file: " + rootMetadataFileName);
- }
-
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Completed the initialization of the local resource repository");
- }
- return;
- }
+ rootMetadataFile.delete();
+ String mountedRootDir;
+ if (rootDir.startsWith(System.getProperty("file.separator"))) {
+ mountedRootDir = new String(mountPoints[i]
+ + rootDir.substring(System.getProperty("file.separator").length()));
+ } else {
+ mountedRootDir = new String(mountPoints[i] + rootDir);
+ }
+ LocalResource rootLocalResource = new LocalResource(ROOT_LOCAL_RESOURCE_ID, rootMetadataFileName, 0, 0,
+ mountedRootDir);
+ insert(rootLocalResource, i);
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("created the root-metadata-file: " + rootMetadataFileName);
+ }
- //#. if the rootMetadataFile exists, read it and set this.rootDir.
- rootLocalResource = readLocalResource(rootMetadataFile);
- this.rootDir = (String) rootLocalResource.getResourceObject();
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("The root directory of the local resource repository is " + this.rootDir);
- }
-
- //#. load all local resources.
- File rootDirFile = new File(this.rootDir);
- if (!rootDirFile.exists()) {
- //rootDir may not exist if this node is not the metadata node and doesn't have any user data.
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("The root directory of the local resource repository doesn't exist: there is no local resource.");
- LOGGER.info("Completed the initialization of the local resource repository");
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Completed the initialization of the local resource repository");
+ }
}
return;
}
@@ -124,27 +116,50 @@
}
};
- long maxResourceId = 0;
- File[] dataverseFileList = rootDirFile.listFiles();
- if (dataverseFileList == null) {
- throw new HyracksDataException("Metadata dataverse doesn't exist.");
- }
- for (File dataverseFile : dataverseFileList) {
- if (dataverseFile.isDirectory()) {
- File[] indexFileList = dataverseFile.listFiles();
- if (indexFileList != null) {
- for (File indexFile : indexFileList) {
- if (indexFile.isDirectory()) {
- File[] metadataFiles = indexFile.listFiles(filter);
- if (metadataFiles != null) {
- for (File metadataFile : metadataFiles) {
- LocalResource localResource = readLocalResource(metadataFile);
- id2ResourceMap.put(localResource.getResourceId(), localResource);
- name2ResourceMap.put(localResource.getResourceName(), localResource);
- maxResourceId = Math.max(localResource.getResourceId(), maxResourceId);
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("loaded local resource - [id: " + localResource.getResourceId()
- + ", name: " + localResource.getResourceName() + "]");
+ for (int i = 0; i < numIODevices; i++) {
+ String rootMetadataFileName = new String(mountPoints[i] + ROOT_METADATA_DIRECTORY
+ + ROOT_METADATA_FILE_NAME_PREFIX + nodeId);
+ File rootMetadataFile = new File(rootMetadataFileName);
+ //#. if the rootMetadataFile exists, read it and set this.rootDir.
+ LocalResource rootLocalResource = readLocalResource(rootMetadataFile);
+ String mountedRootDir = (String) rootLocalResource.getResourceObject();
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("The root directory of the local resource repository is " + mountedRootDir);
+ }
+
+ //#. load all local resources.
+ File rootDirFile = new File(mountedRootDir);
+ if (!rootDirFile.exists()) {
+ //rootDir may not exist if this node is not the metadata node and doesn't have any user data.
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("The root directory of the local resource repository doesn't exist: there is no local resource.");
+ LOGGER.info("Completed the initialization of the local resource repository");
+ }
+ continue;
+ }
+
+ long maxResourceId = 0;
+ File[] dataverseFileList = rootDirFile.listFiles();
+ if (dataverseFileList == null) {
+ throw new HyracksDataException("Metadata dataverse doesn't exist.");
+ }
+ for (File dataverseFile : dataverseFileList) {
+ if (dataverseFile.isDirectory()) {
+ File[] indexFileList = dataverseFile.listFiles();
+ if (indexFileList != null) {
+ for (File indexFile : indexFileList) {
+ if (indexFile.isDirectory()) {
+ File[] metadataFiles = indexFile.listFiles(filter);
+ if (metadataFiles != null) {
+ for (File metadataFile : metadataFiles) {
+ LocalResource localResource = readLocalResource(metadataFile);
+ id2ResourceMap.put(localResource.getResourceId(), localResource);
+ name2ResourceMap.put(localResource.getResourceName(), localResource);
+ maxResourceId = Math.max(localResource.getResourceId(), maxResourceId);
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("loaded local resource - [id: " + localResource.getResourceId()
+ + ", name: " + localResource.getResourceName() + "]");
+ }
}
}
}
@@ -152,11 +167,11 @@
}
}
}
- }
- resourceIdFactory.initId(maxResourceId + 1);
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("The resource id factory is intialized with the value: " + (maxResourceId + 1));
- LOGGER.info("Completed the initialization of the local resource repository");
+ resourceIdFactory.initId(maxResourceId + 1);
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("The resource id factory is intialized with the value: " + (maxResourceId + 1));
+ LOGGER.info("Completed the initialization of the local resource repository");
+ }
}
}
@@ -171,7 +186,7 @@
}
@Override
- public synchronized void insert(LocalResource resource) throws HyracksDataException {
+ public synchronized void insert(LocalResource resource, int ioDeviceId) throws HyracksDataException {
long id = resource.getResourceId();
if (id2ResourceMap.containsKey(id)) {
@@ -185,8 +200,10 @@
FileOutputStream fos = null;
ObjectOutputStream oosToFos = null;
+
try {
- fos = new FileOutputStream(getFileName(mountPoint, resource.getResourceName(), resource.getResourceId()));
+ fos = new FileOutputStream(getFileName(mountPoints[ioDeviceId], resource.getResourceName(),
+ resource.getResourceId()));
oosToFos = new ObjectOutputStream(fos);
oosToFos.writeObject(resource);
oosToFos.flush();
@@ -211,26 +228,26 @@
}
@Override
- public synchronized void deleteResourceById(long id) throws HyracksDataException {
+ public synchronized void deleteResourceById(long id, int ioDeviceId) throws HyracksDataException {
LocalResource resource = id2ResourceMap.get(id);
if (resource == null) {
throw new HyracksDataException("Resource doesn't exist");
}
id2ResourceMap.remove(id);
name2ResourceMap.remove(resource.getResourceName());
- File file = new File(getFileName(mountPoint, resource.getResourceName(), resource.getResourceId()));
+ File file = new File(getFileName(mountPoints[ioDeviceId], resource.getResourceName(), resource.getResourceId()));
file.delete();
}
@Override
- public synchronized void deleteResourceByName(String name) throws HyracksDataException {
+ public synchronized void deleteResourceByName(String name, int ioDeviceId) throws HyracksDataException {
LocalResource resource = name2ResourceMap.get(name);
if (resource == null) {
throw new HyracksDataException("Resource doesn't exist");
}
id2ResourceMap.remove(resource.getResourceId());
name2ResourceMap.remove(name);
- File file = new File(getFileName(mountPoint, resource.getResourceName(), resource.getResourceId()));
+ File file = new File(getFileName(mountPoints[ioDeviceId], resource.getResourceName(), resource.getResourceId()));
file.delete();
}
diff --git a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
index c7efca5..f6847f9 100644
--- a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
+++ b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
@@ -14,11 +14,8 @@
*/
package edu.uci.ics.asterix.transaction.management.resource;
-import java.util.List;
-
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.io.IIOManager;
-import edu.uci.ics.hyracks.api.io.IODeviceHandle;
import edu.uci.ics.hyracks.storage.common.file.ILocalResourceRepository;
import edu.uci.ics.hyracks.storage.common.file.ILocalResourceRepositoryFactory;
@@ -31,7 +28,6 @@
@Override
public ILocalResourceRepository createRepository() throws HyracksDataException {
- List<IODeviceHandle> devices = ioManager.getIODevices();
- return new PersistentLocalResourceRepository(devices.get(0).getPath().getPath());
+ return new PersistentLocalResourceRepository(ioManager.getIODevices());
}
-}
+}
\ No newline at end of file