1) Use of absolute location constraints: HDFS Adapter will attempt to form absolute location constraints from the information retreived from the Namenode. If for any input split, there is no node controller running at the location, teh adapter will fall back to using count constraints.
2) Asterix Bootstrap: During bootstrap, all information regarding the location of the node controllers is collected and is made available for use later.
3) Re-shuffling of code: Moved classes to asterix-common.
git-svn-id: https://asterixdb.googlecode.com/svn/branches/asterix_stabilization@165 eaa15691-b419-025a-1212-ee371bd00084
diff --git a/asterix-app/pom.xml b/asterix-app/pom.xml
index 430d8a5..5d7a187 100644
--- a/asterix-app/pom.xml
+++ b/asterix-app/pom.xml
@@ -170,6 +170,13 @@
<scope>compile</scope>
</dependency>
<dependency>
+ <groupId>edu.uci.ics.asterix</groupId>
+ <artifactId>asterix-common</artifactId>
+ <version>0.0.4-SNAPSHOT</version>
+ <type>jar</type>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
<groupId>com.kenai.nbpwr</groupId>
<artifactId>org-apache-commons-io</artifactId>
<version>1.3.1-201002241208</version>
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/APIFramework.java b/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/APIFramework.java
index 85b2b87..b0b27ed 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/APIFramework.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/APIFramework.java
@@ -4,6 +4,7 @@
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import org.json.JSONException;
@@ -13,6 +14,7 @@
import edu.uci.ics.asterix.aql.expression.visitor.AQLPrintVisitor;
import edu.uci.ics.asterix.aql.rewrites.AqlRewriter;
import edu.uci.ics.asterix.aql.translator.DdlTranslator;
+import edu.uci.ics.asterix.common.api.AsterixAppContextInfoImpl;
import edu.uci.ics.asterix.common.config.GlobalConfig;
import edu.uci.ics.asterix.common.config.OptimizationConfUtil;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
@@ -67,6 +69,7 @@
import edu.uci.ics.hyracks.algebricks.core.rewriter.base.PhysicalOptimizationConfig;
import edu.uci.ics.hyracks.algebricks.core.utils.Pair;
import edu.uci.ics.hyracks.api.client.IHyracksClientConnection;
+import edu.uci.ics.hyracks.api.client.NodeControllerInfo;
import edu.uci.ics.hyracks.api.job.JobId;
import edu.uci.ics.hyracks.api.job.JobSpecification;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/AsterixAppContextInfoImpl.java b/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/AsterixAppContextInfoImpl.java
deleted file mode 100644
index 5432fb9..0000000
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/api/common/AsterixAppContextInfoImpl.java
+++ /dev/null
@@ -1,27 +0,0 @@
-package edu.uci.ics.asterix.api.common;
-
-import edu.uci.ics.asterix.context.AsterixStorageManagerInterface;
-import edu.uci.ics.asterix.context.AsterixTreeRegistryProvider;
-import edu.uci.ics.asterix.dataflow.base.IAsterixApplicationContextInfo;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
-import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
-
-public class AsterixAppContextInfoImpl implements IAsterixApplicationContextInfo {
-
- public static final AsterixAppContextInfoImpl INSTANCE = new AsterixAppContextInfoImpl();
-
- private AsterixAppContextInfoImpl() {
- }
-
- @Override
- public IIndexRegistryProvider<IIndex> getTreeRegisterProvider() {
- return AsterixTreeRegistryProvider.INSTANCE;
- }
-
- @Override
- public IStorageManagerInterface getStorageManagerInterface() {
- return AsterixStorageManagerInterface.INSTANCE;
- }
-
-}
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 3666251..9f41b65 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
@@ -21,12 +21,12 @@
import edu.uci.ics.asterix.api.common.Job;
import edu.uci.ics.asterix.aql.translator.DdlTranslator.CompiledDatasetDropStatement;
-import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
import edu.uci.ics.asterix.common.config.GlobalConfig;
import edu.uci.ics.asterix.common.config.OptimizationConfUtil;
+import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
+import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
+import edu.uci.ics.asterix.common.context.AsterixTreeRegistryProvider;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.context.AsterixStorageManagerInterface;
-import edu.uci.ics.asterix.context.AsterixTreeRegistryProvider;
import edu.uci.ics.asterix.formats.base.IDataFormat;
import edu.uci.ics.asterix.metadata.MetadataException;
import edu.uci.ics.asterix.metadata.declared.AqlCompiledDatasetDecl;
@@ -44,7 +44,6 @@
import edu.uci.ics.hyracks.algebricks.core.algebra.runtime.base.IEvaluatorFactory;
import edu.uci.ics.hyracks.algebricks.core.algebra.runtime.base.IPushRuntimeFactory;
import edu.uci.ics.hyracks.algebricks.core.algebra.runtime.jobgen.impl.ConnectorPolicyAssignmentPolicy;
-import edu.uci.ics.hyracks.algebricks.core.algebra.runtime.jobgen.impl.JobGenHelper;
import edu.uci.ics.hyracks.algebricks.core.algebra.runtime.operators.meta.AlgebricksMetaOperatorDescriptor;
import edu.uci.ics.hyracks.algebricks.core.algebra.runtime.operators.std.AssignRuntimeFactory;
import edu.uci.ics.hyracks.algebricks.core.api.constraints.AlgebricksPartitionConstraint;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/IndexOperations.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/IndexOperations.java
index 5e6ffa1..730fac6 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/IndexOperations.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/IndexOperations.java
@@ -7,9 +7,9 @@
import edu.uci.ics.asterix.aql.translator.DdlTranslator.CompiledIndexDropStatement;
import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
import edu.uci.ics.asterix.common.config.OptimizationConfUtil;
+import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
+import edu.uci.ics.asterix.common.context.AsterixTreeRegistryProvider;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.context.AsterixStorageManagerInterface;
-import edu.uci.ics.asterix.context.AsterixTreeRegistryProvider;
import edu.uci.ics.asterix.dataflow.data.nontagged.valueproviders.AqlPrimitiveValueProviderFactory;
import edu.uci.ics.asterix.formats.nontagged.AqlBinaryComparatorFactoryProvider;
import edu.uci.ics.asterix.formats.nontagged.AqlBinaryTokenizerFactoryProvider;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestKeywordIndexJob.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestKeywordIndexJob.java
index 91af8de..a4a84db 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestKeywordIndexJob.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestKeywordIndexJob.java
@@ -6,9 +6,9 @@
import java.util.HashMap;
import java.util.List;
+import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
+import edu.uci.ics.asterix.common.context.AsterixTreeRegistryProvider;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.context.AsterixStorageManagerInterface;
-import edu.uci.ics.asterix.context.AsterixTreeRegistryProvider;
import edu.uci.ics.asterix.dataflow.data.nontagged.comparators.AObjectAscBinaryComparatorFactory;
import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AObjectSerializerDeserializer;
import edu.uci.ics.asterix.metadata.declared.AqlMetadataProvider;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestSecondaryIndexJob.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestSecondaryIndexJob.java
index 025fa16..cdecd74 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestSecondaryIndexJob.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestSecondaryIndexJob.java
@@ -5,18 +5,16 @@
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
-import java.util.UUID;
+import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
+import edu.uci.ics.asterix.common.context.AsterixTreeRegistryProvider;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.context.AsterixStorageManagerInterface;
-import edu.uci.ics.asterix.context.AsterixTreeRegistryProvider;
import edu.uci.ics.asterix.dataflow.data.nontagged.comparators.AObjectAscBinaryComparatorFactory;
import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AObjectSerializerDeserializer;
import edu.uci.ics.asterix.metadata.declared.AqlMetadataProvider;
import edu.uci.ics.asterix.om.base.AString;
import edu.uci.ics.asterix.om.types.ATypeTag;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
-import edu.uci.ics.hyracks.algebricks.core.algebra.runtime.jobgen.impl.JobGenHelper;
import edu.uci.ics.hyracks.api.client.HyracksConnection;
import edu.uci.ics.hyracks.api.client.IHyracksClientConnection;
import edu.uci.ics.hyracks.api.constraints.PartitionConstraintHelper;
@@ -43,7 +41,6 @@
import edu.uci.ics.hyracks.dataflow.std.misc.PrinterOperatorDescriptor;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeDataflowHelperFactory;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/CCBootstrapImpl.java b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/CCBootstrapImpl.java
index bcbfb8c..c6d38d3 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/CCBootstrapImpl.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/CCBootstrapImpl.java
@@ -28,6 +28,7 @@
import edu.uci.ics.asterix.api.aqlj.server.APIClientThreadFactory;
import edu.uci.ics.asterix.api.aqlj.server.ThreadedServer;
import edu.uci.ics.asterix.api.http.servlet.APIServlet;
+import edu.uci.ics.asterix.common.api.AsterixAppContextInfoImpl;
import edu.uci.ics.asterix.common.config.GlobalConfig;
import edu.uci.ics.asterix.metadata.MetadataManager;
import edu.uci.ics.asterix.metadata.api.IAsterixStateProxy;
@@ -74,6 +75,7 @@
} catch (Exception e) {
throw new IOException(" unable to obtain IP address node map", e);
}
+ AsterixAppContextInfoImpl.setNodeControllerInfo(nodeNameMap);
for (Map.Entry<String, Set<String>> entry : nodeNameMap.entrySet()) {
Set<String> nodeNames = entry.getValue();
Iterator<String> it = nodeNames.iterator();
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/NCBootstrapImpl.java b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/NCBootstrapImpl.java
index 63fb497..7cdca49 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/NCBootstrapImpl.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/NCBootstrapImpl.java
@@ -19,9 +19,9 @@
import edu.uci.ics.asterix.api.aqlj.server.NodeDataClientThreadFactory;
import edu.uci.ics.asterix.api.aqlj.server.ThreadedServer;
-import edu.uci.ics.asterix.api.common.AsterixAppContextInfoImpl;
+import edu.uci.ics.asterix.common.api.AsterixAppContextInfoImpl;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.context.AsterixAppRuntimeContext;
+import edu.uci.ics.asterix.common.context.AsterixAppRuntimeContext;
import edu.uci.ics.asterix.metadata.MetadataManager;
import edu.uci.ics.asterix.metadata.MetadataNode;
import edu.uci.ics.asterix.metadata.api.IAsterixStateProxy;
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/api/AsterixAppContextInfoImpl.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/api/AsterixAppContextInfoImpl.java
new file mode 100644
index 0000000..39642d1
--- /dev/null
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/api/AsterixAppContextInfoImpl.java
@@ -0,0 +1,40 @@
+package edu.uci.ics.asterix.common.api;
+
+import java.util.Map;
+import java.util.Set;
+
+import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
+import edu.uci.ics.asterix.common.context.AsterixTreeRegistryProvider;
+import edu.uci.ics.asterix.common.dataflow.IAsterixApplicationContextInfo;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
+
+public class AsterixAppContextInfoImpl implements IAsterixApplicationContextInfo {
+
+ public static final AsterixAppContextInfoImpl INSTANCE = new AsterixAppContextInfoImpl();
+
+ private static Map<String, Set<String>> nodeControllerMap;
+
+ private AsterixAppContextInfoImpl() {
+ }
+
+ @Override
+ public IIndexRegistryProvider<IIndex> getTreeRegisterProvider() {
+ return AsterixTreeRegistryProvider.INSTANCE;
+ }
+
+ @Override
+ public IStorageManagerInterface getStorageManagerInterface() {
+ return AsterixStorageManagerInterface.INSTANCE;
+ }
+
+ public static void setNodeControllerInfo(Map<String, Set<String>> nodeControllerInfo) {
+ nodeControllerMap = nodeControllerInfo;
+ }
+
+ public static Map<String, Set<String>> getNodeControllerMap() {
+ return nodeControllerMap;
+ }
+
+}
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixAppRuntimeContext.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixAppRuntimeContext.java
similarity index 98%
rename from asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixAppRuntimeContext.java
rename to asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixAppRuntimeContext.java
index 6320737..949855f 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixAppRuntimeContext.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixAppRuntimeContext.java
@@ -1,4 +1,4 @@
-package edu.uci.ics.asterix.context;
+package edu.uci.ics.asterix.common.context;
import java.io.IOException;
import java.io.PrintWriter;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixFileMapManager.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixFileMapManager.java
similarity index 97%
rename from asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixFileMapManager.java
rename to asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixFileMapManager.java
index 0aacc3d..5c6611b 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixFileMapManager.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixFileMapManager.java
@@ -1,4 +1,4 @@
-package edu.uci.ics.asterix.context;
+package edu.uci.ics.asterix.common.context;
import java.io.File;
import java.util.HashMap;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixStorageManagerInterface.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixStorageManagerInterface.java
similarity index 94%
rename from asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixStorageManagerInterface.java
rename to asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixStorageManagerInterface.java
index 12d715d..67b785e 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixStorageManagerInterface.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixStorageManagerInterface.java
@@ -1,4 +1,4 @@
-package edu.uci.ics.asterix.context;
+package edu.uci.ics.asterix.common.context;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixTreeRegistryProvider.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixTreeRegistryProvider.java
similarity index 93%
rename from asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixTreeRegistryProvider.java
rename to asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixTreeRegistryProvider.java
index 80fd64a..3f542ec 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/context/AsterixTreeRegistryProvider.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixTreeRegistryProvider.java
@@ -1,4 +1,4 @@
-package edu.uci.ics.asterix.context;
+package edu.uci.ics.asterix.common.context;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/base/IAsterixApplicationContextInfo.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java
similarity index 89%
rename from asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/base/IAsterixApplicationContextInfo.java
rename to asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java
index d8d9e78..b5033d3 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/base/IAsterixApplicationContextInfo.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java
@@ -1,4 +1,4 @@
-package edu.uci.ics.asterix.dataflow.base;
+package edu.uci.ics.asterix.common.dataflow;
import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/data/adapter/api/IDatasourceAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/data/adapter/api/IDatasourceAdapter.java
index 1baf149..ea21d78 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/data/adapter/api/IDatasourceAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/data/adapter/api/IDatasourceAdapter.java
@@ -19,6 +19,7 @@
import edu.uci.ics.asterix.om.types.IAType;
import edu.uci.ics.hyracks.algebricks.core.api.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.api.client.NodeControllerInfo;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
@@ -143,7 +144,8 @@
* providing all arguments as a set of (key,value) pairs. These
* arguments are put into the metadata.
*/
- public void configure(Map<String, String> arguments, IAType atype) throws Exception;
+ public void configure(Map<String, String> arguments, IAType atype)
+ throws Exception;
/**
* Returns a list of partition constraints. A partition constraint can be a
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HDFSAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HDFSAdapter.java
index f7c89ff..6a0d181 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HDFSAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HDFSAdapter.java
@@ -17,11 +17,16 @@
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
+import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.logging.Level;
+import java.util.logging.Logger;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -42,6 +47,8 @@
import edu.uci.ics.asterix.om.types.IAType;
import edu.uci.ics.asterix.runtime.operators.file.AdmSchemafullRecordParserFactory;
import edu.uci.ics.asterix.runtime.operators.file.NtDelimitedDataTupleParserFactory;
+import edu.uci.ics.asterix.runtime.util.AsterixRuntimeUtil;
+import edu.uci.ics.hyracks.algebricks.core.api.constraints.AlgebricksAbsolutePartitionConstraint;
import edu.uci.ics.hyracks.algebricks.core.api.constraints.AlgebricksCountPartitionConstraint;
import edu.uci.ics.hyracks.algebricks.core.api.exceptions.NotImplementedException;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
@@ -52,8 +59,8 @@
public class HDFSAdapter extends AbstractDatasourceAdapter implements IDatasourceReadAdapter {
- private String hdfsUrl;
- private List<String> hdfsPaths;
+ private static final Logger LOGGER = Logger.getLogger(HDFSAdapter.class.getName());
+
private String inputFormatClassName;
private Object[] inputSplits;
private transient JobConf conf;
@@ -76,22 +83,6 @@
formatClassNames.put(INPUT_FORMAT_SEQUENCE, "org.apache.hadoop.mapred.SequenceFileInputFormat");
}
- public String getHdfsUrl() {
- return hdfsUrl;
- }
-
- public void setHdfsUrl(String hdfsUrl) {
- this.hdfsUrl = hdfsUrl;
- }
-
- public List<String> getHdfsPaths() {
- return hdfsPaths;
- }
-
- public void setHdfsPaths(List<String> hdfsPaths) {
- this.hdfsPaths = hdfsPaths;
- }
-
@Override
public void configure(Map<String, String> arguments, IAType atype) throws Exception {
configuration = arguments;
@@ -126,14 +117,52 @@
}
private void configurePartitionConstraint() throws Exception {
+ AlgebricksAbsolutePartitionConstraint absPartitionConstraint;
+ List<String> locations = new ArrayList<String>();
+ Random random = new Random();
+ boolean couldConfigureLocationConstraints = true;
if (inputSplitsProxy == null) {
InputSplit[] inputSplits = conf.getInputFormat().getSplits(conf, 0);
- inputSplitsProxy = new InputSplitsProxy(conf, inputSplits);
- partitionConstraint = new AlgebricksCountPartitionConstraint(inputSplits.length);
- hdfsPaths = new ArrayList<String>();
- for (String hdfsPath : configuration.get(KEY_HDFS_PATH).split(",")) {
- hdfsPaths.add(hdfsPath);
+ try {
+ for (InputSplit inputSplit : inputSplits) {
+ String[] dataNodeLocations = inputSplit.getLocations();
+ for (String datanodeLocation : dataNodeLocations) {
+ Set<String> nodeControllersAtLocation = AsterixRuntimeUtil
+ .getNodeControllersOnHostName(datanodeLocation);
+ if (nodeControllersAtLocation == null || nodeControllersAtLocation.size() == 0) {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.log(Level.INFO, "No node controller found at " + datanodeLocation
+ + " will look at replica location");
+ }
+ couldConfigureLocationConstraints = false;
+ } else {
+ int locationIndex = random.nextInt(nodeControllersAtLocation.size());
+ String chosenLocation = (String) nodeControllersAtLocation.toArray()[locationIndex];
+ locations.add(chosenLocation);
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.log(Level.INFO, "split : " + inputSplit + " to be processed by :"
+ + chosenLocation);
+ }
+ couldConfigureLocationConstraints = true;
+ break;
+ }
+ }
+ if(!couldConfigureLocationConstraints){
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.log(Level.INFO, "No local node controller found to process split : " + inputSplit + " will use count constraint!");
+ }
+ break;
+ }
+ }
+ if (couldConfigureLocationConstraints) {
+ partitionConstraint = new AlgebricksAbsolutePartitionConstraint(locations.toArray(new String[] {}));
+ } else {
+ partitionConstraint = new AlgebricksCountPartitionConstraint(inputSplits.length);
+ }
+ } catch (UnknownHostException e) {
+ partitionConstraint = new AlgebricksCountPartitionConstraint(inputSplits.length);
}
+ inputSplitsProxy = new InputSplitsProxy(conf, inputSplits);
}
}
@@ -156,9 +185,8 @@
}
private JobConf configureJobConf() throws Exception {
- hdfsUrl = configuration.get(KEY_HDFS_URL);
conf = new JobConf();
- conf.set("fs.default.name", hdfsUrl);
+ conf.set("fs.default.name", configuration.get(KEY_HDFS_URL));
conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
conf.setClassLoader(HDFSAdapter.class.getClassLoader());
conf.set("mapred.input.dir", configuration.get(KEY_HDFS_PATH));
@@ -214,10 +242,10 @@
public void progress() {
}
};
-
+
return reporter;
}
-
+
@Override
public IDataParser getDataParser(int partition) throws Exception {
Path path = new Path(inputSplits[partition].toString());
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java
index dabff59..eb73938 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java
@@ -21,8 +21,8 @@
import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
import edu.uci.ics.asterix.common.config.DatasetConfig.IndexType;
+import edu.uci.ics.asterix.common.dataflow.IAsterixApplicationContextInfo;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.dataflow.base.IAsterixApplicationContextInfo;
import edu.uci.ics.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
import edu.uci.ics.asterix.metadata.api.IMetadataIndex;
import edu.uci.ics.asterix.metadata.api.IMetadataNode;
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 29df060..84fb721 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
@@ -25,7 +25,7 @@
import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
import edu.uci.ics.asterix.common.config.DatasetConfig.IndexType;
-import edu.uci.ics.asterix.dataflow.base.IAsterixApplicationContextInfo;
+import edu.uci.ics.asterix.common.dataflow.IAsterixApplicationContextInfo;
import edu.uci.ics.asterix.metadata.IDatasetDetails;
import edu.uci.ics.asterix.metadata.MetadataManager;
import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
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 17deaaf..f99d4ff 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
@@ -17,11 +17,12 @@
import java.io.File;
import java.util.List;
+import java.util.Map;
import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
import edu.uci.ics.asterix.common.config.GlobalConfig;
+import edu.uci.ics.asterix.common.dataflow.IAsterixApplicationContextInfo;
import edu.uci.ics.asterix.common.parse.IParseFileSplitsDecl;
-import edu.uci.ics.asterix.dataflow.base.IAsterixApplicationContextInfo;
import edu.uci.ics.asterix.dataflow.data.nontagged.valueproviders.AqlPrimitiveValueProviderFactory;
import edu.uci.ics.asterix.external.data.adapter.api.IDatasourceAdapter;
import edu.uci.ics.asterix.external.data.adapter.api.IDatasourceReadAdapter;
@@ -62,6 +63,7 @@
import edu.uci.ics.hyracks.algebricks.core.api.exceptions.AlgebricksException;
import edu.uci.ics.hyracks.algebricks.core.utils.Pair;
import edu.uci.ics.hyracks.algebricks.core.utils.Triple;
+import edu.uci.ics.hyracks.api.client.NodeControllerInfo;
import edu.uci.ics.hyracks.api.dataflow.IOperatorDescriptor;
import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
@@ -217,9 +219,10 @@
.getAdapterType().equals(
IDatasourceAdapter.AdapterType.READ_WRITE))) {
throw new AlgebricksException(
- "external dataset does not support read");
+ "external dataset adapter does not support read operation");
}
ARecordType rt = (ARecordType) itemType;
+
try {
adapter.configure(decl.getProperties(), itemType);
} catch (Exception e) {
@@ -291,6 +294,7 @@
}
ARecordType rt = (ARecordType) itemType;
+ Map<String, NodeControllerInfo> ncInfo = null;
try {
adapter.configure(decl.getProperties(), itemType);
} catch (Exception e) {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/AsterixRuntimeUtil.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/AsterixRuntimeUtil.java
new file mode 100644
index 0000000..5326c56
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/AsterixRuntimeUtil.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2009-2011 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.runtime.util;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Map;
+import java.util.Set;
+
+import edu.uci.ics.asterix.common.api.AsterixAppContextInfoImpl;
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+
+public class AsterixRuntimeUtil {
+
+ public static Set<String> getNodeControllersOnIP(String ipAddress) throws AsterixException {
+ Map<String, Set<String>> nodeControllerInfo = AsterixAppContextInfoImpl.getNodeControllerMap();
+ Set<String> nodeControllersAtLocation = nodeControllerInfo.get(ipAddress);
+ return nodeControllersAtLocation;
+ }
+
+ public static Set<String> getNodeControllersOnHostName(String hostName) throws UnknownHostException {
+ Map<String, Set<String>> nodeControllerInfo = AsterixAppContextInfoImpl.getNodeControllerMap();
+ String address;
+ address = InetAddress.getByName(hostName).getHostAddress();
+ if (address.equals("127.0.1.1")) {
+ address = "127.0.0.1";
+ }
+ Set<String> nodeControllersAtLocation = nodeControllerInfo.get(address);
+ return nodeControllersAtLocation;
+ }
+
+}