moved work related to external UDFs/libraries to descendant of asterix_lsm_stabilization from a descendant of asterix_stabilization
git-svn-id: https://asterixdb.googlecode.com/svn/branches/asterix_lsm_stabilization_udfs@1719 eaa15691-b419-025a-1212-ee371bd00084
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/jobgen/AqlLogicalExpressionJobGen.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/jobgen/AqlLogicalExpressionJobGen.java
index dd8791f..8523e75 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/jobgen/AqlLogicalExpressionJobGen.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/jobgen/AqlLogicalExpressionJobGen.java
@@ -4,10 +4,13 @@
import org.apache.commons.lang3.mutable.Mutable;
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
import edu.uci.ics.asterix.common.functions.FunctionDescriptorTag;
+import edu.uci.ics.asterix.external.library.ExternalFunctionDescriptorProvider;
import edu.uci.ics.asterix.formats.base.IDataFormat;
import edu.uci.ics.asterix.metadata.declared.AqlMetadataProvider;
import edu.uci.ics.asterix.om.functions.AsterixBuiltinFunctions;
+import edu.uci.ics.asterix.om.functions.IExternalFunctionInfo;
import edu.uci.ics.asterix.om.functions.IFunctionDescriptor;
import edu.uci.ics.asterix.runtime.evaluators.comparisons.ComparisonEvalFactory;
import edu.uci.ics.asterix.runtime.formats.FormatUtils;
@@ -124,9 +127,18 @@
}
IFunctionDescriptor fd = null;
- AqlMetadataProvider mp = (AqlMetadataProvider) context.getMetadataProvider();
- IDataFormat format = FormatUtils.getDefaultFormat();
- fd = format.resolveFunction(expr, env);
+ if (!(expr.getFunctionInfo() instanceof IExternalFunctionInfo)) {
+ AqlMetadataProvider mp = (AqlMetadataProvider) context.getMetadataProvider();
+ IDataFormat format = FormatUtils.getDefaultFormat();
+ fd = format.resolveFunction(expr, env);
+ } else {
+ try {
+ fd = ExternalFunctionDescriptorProvider.getExternalFunctionDescriptor((IExternalFunctionInfo) expr
+ .getFunctionInfo());
+ } catch (AsterixException ae) {
+ throw new AlgebricksException(ae);
+ }
+ }
return fd.createEvaluatorFactory(args);
}
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AqlExpressionToPlanTranslator.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AqlExpressionToPlanTranslator.java
index caa16d4..0a5e4c5 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AqlExpressionToPlanTranslator.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AqlExpressionToPlanTranslator.java
@@ -87,6 +87,7 @@
import edu.uci.ics.asterix.metadata.declared.ResultSetSinkId;
import edu.uci.ics.asterix.metadata.entities.Dataset;
import edu.uci.ics.asterix.metadata.entities.Function;
+import edu.uci.ics.asterix.metadata.functions.ExternalFunctionCompilerUtil;
import edu.uci.ics.asterix.metadata.utils.DatasetUtils;
import edu.uci.ics.asterix.om.base.AInt32;
import edu.uci.ics.asterix.om.base.AString;
@@ -494,13 +495,18 @@
return null;
}
AbstractFunctionCallExpression f = null;
- if (function.getLanguage().equalsIgnoreCase(Function.LANGUAGE_AQL)) {
+ if (function.getLanguage().equalsIgnoreCase(Function.LANGUAGE_JAVA)) {
+ IFunctionInfo finfo = ExternalFunctionCompilerUtil.getExternalFunctionInfo(
+ metadataProvider.getMetadataTxnContext(), function);
+ f = new ScalarFunctionCallExpression(finfo, args);
+ } else if (function.getLanguage().equalsIgnoreCase(Function.LANGUAGE_AQL)) {
IFunctionInfo finfo = new AsterixFunctionInfo(signature);
- return new ScalarFunctionCallExpression(finfo, args);
+ f = new ScalarFunctionCallExpression(finfo, args);
} else {
throw new MetadataException(" User defined functions written in " + function.getLanguage()
+ " are not supported");
}
+ return f;
}
private AbstractFunctionCallExpression lookupBuiltinFunction(String functionName, int arity,
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 5ab94f3..c8ed425 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
@@ -59,6 +59,7 @@
jsonAPIServer.start();
AsterixAppContextInfoImpl.initialize(appCtx);
+ ExternalLibraryBootstrap.setUpExternaLibraries(false);
}
@Override
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/ExternalLibraryBootstrap.java b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/ExternalLibraryBootstrap.java
new file mode 100755
index 0000000..3b214c6
--- /dev/null
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/ExternalLibraryBootstrap.java
@@ -0,0 +1,300 @@
+/*
+ * Copyright 2009-2012 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.io.File;
+import java.io.FilenameFilter;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.rmi.RemoteException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.xml.bind.JAXBContext;
+import javax.xml.bind.Unmarshaller;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.common.functions.FunctionSignature;
+import edu.uci.ics.asterix.external.library.ExternalLibraryManager;
+import edu.uci.ics.asterix.external.library.Function;
+import edu.uci.ics.asterix.external.library.Functions;
+import edu.uci.ics.asterix.external.library.Library;
+import edu.uci.ics.asterix.metadata.MetadataManager;
+import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
+import edu.uci.ics.asterix.metadata.api.IMetadataEntity;
+import edu.uci.ics.asterix.metadata.entities.Dataverse;
+import edu.uci.ics.asterix.runtime.formats.NonTaggedDataFormat;
+import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
+
+public class ExternalLibraryBootstrap {
+
+ public static void setUpExternaLibraries(boolean isMetadataNode) throws Exception {
+
+ Map<String, List<String>> uninstalledLibs = null;
+ if (isMetadataNode) {
+ uninstalledLibs = uninstallLibraries();
+ }
+
+ File installLibDir = getLibraryInstallDir();
+ if (installLibDir.exists()) {
+ for (String dataverse : installLibDir.list()) {
+ File dataverseDir = new File(installLibDir, dataverse);
+ String[] libraries = dataverseDir.list();
+ for (String library : libraries) {
+ registerLibrary(dataverse, library, isMetadataNode, installLibDir);
+ if (isMetadataNode) {
+ File libraryDir = new File(installLibDir.getAbsolutePath() + File.separator + dataverse
+ + File.separator + library);
+ installLibraryIfNeeded(dataverse, libraryDir, uninstalledLibs);
+ }
+ }
+ }
+ }
+ }
+
+ private static Map<String, List<String>> uninstallLibraries() throws Exception {
+ Map<String, List<String>> uninstalledLibs = new HashMap<String, List<String>>();
+ File uninstallLibDir = getLibraryUninstallDir();
+ String[] uninstallLibNames;
+ if (uninstallLibDir.exists()) {
+ uninstallLibNames = uninstallLibDir.list();
+ for (String uninstallLibName : uninstallLibNames) {
+ String[] components = uninstallLibName.split("\\.");
+ String dataverse = components[0];
+ String libName = components[1];
+ uninstallLibrary(dataverse, libName);
+ new File(uninstallLibDir, uninstallLibName).delete();
+ List<String> uinstalledLibsInDv = uninstalledLibs.get(dataverse);
+ if (uinstalledLibsInDv == null) {
+ uinstalledLibsInDv = new ArrayList<String>();
+ uninstalledLibs.put(dataverse, uinstalledLibsInDv);
+ }
+ uinstalledLibsInDv.add(libName);
+ }
+ }
+ return uninstalledLibs;
+ }
+
+ private static boolean uninstallLibrary(String dataverse, String libraryName) throws AsterixException,
+ RemoteException, ACIDException {
+ MetadataTransactionContext mdTxnCtx = null;
+ try {
+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
+ Dataverse dv = MetadataManager.INSTANCE.getDataverse(mdTxnCtx, dataverse);
+ if (dv == null) {
+ return false;
+ }
+
+ edu.uci.ics.asterix.metadata.entities.Library library = MetadataManager.INSTANCE.getLibrary(mdTxnCtx,
+ dataverse, libraryName);
+ if (library == null) {
+ return false;
+ }
+
+ List<edu.uci.ics.asterix.metadata.entities.Function> functions = MetadataManager.INSTANCE
+ .getDataverseFunctions(mdTxnCtx, dataverse);
+ for (edu.uci.ics.asterix.metadata.entities.Function function : functions) {
+ if (function.getName().startsWith(libraryName + ":")) {
+ MetadataManager.INSTANCE.dropFunction(mdTxnCtx, new FunctionSignature(dataverse,
+ function.getName(), function.getArity()));
+ }
+ }
+
+ MetadataManager.INSTANCE.dropLibrary(mdTxnCtx, dataverse, libraryName);
+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
+ } catch (Exception e) {
+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
+ throw new AsterixException(e);
+ }
+ return true;
+ }
+
+ // Each element of a library is installed as part of a transaction. Any
+ // failure in installing an element does not effect installation of other
+ // libraries
+ private static void installLibraryIfNeeded(String dataverse, final File libraryDir,
+ Map<String, List<String>> uninstalledLibs) throws Exception {
+
+ String libraryName = libraryDir.getName();
+ List<String> uninstalledLibsInDv = uninstalledLibs.get(dataverse);
+ boolean wasUninstalled = uninstalledLibsInDv != null && uninstalledLibsInDv.contains(libraryName);
+
+ MetadataTransactionContext mdTxnCtx = null;
+ try {
+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
+ edu.uci.ics.asterix.metadata.entities.Library libraryInMetadata = MetadataManager.INSTANCE.getLibrary(
+ mdTxnCtx, dataverse, libraryName);
+ if (libraryInMetadata != null && !wasUninstalled) {
+ return;
+ }
+
+ String[] libraryDescriptors = libraryDir.list(new FilenameFilter() {
+ @Override
+ public boolean accept(File dir, String name) {
+ return name.endsWith(".xml");
+ }
+ });
+
+ Library library = getLibrary(new File(libraryDir + File.separator + libraryDescriptors[0]));
+
+ if (libraryDescriptors.length == 0) {
+ throw new Exception("No library descriptors defined");
+ } else if (libraryDescriptors.length > 1) {
+ throw new Exception("More than 1 library descriptors defined");
+ }
+
+ Dataverse dv = MetadataManager.INSTANCE.getDataverse(mdTxnCtx, dataverse);
+ if (dv == null) {
+ MetadataManager.INSTANCE.addDataverse(mdTxnCtx, new Dataverse(dataverse,
+ NonTaggedDataFormat.NON_TAGGED_DATA_FORMAT, IMetadataEntity.PENDING_NO_OP));
+ }
+ for (Function function : library.getFunctions().getFunction()) {
+ String[] fargs = function.getArguments().trim().split(",");
+ List<String> args = new ArrayList<String>();
+ for (String arg : fargs) {
+ args.add(arg);
+ }
+ edu.uci.ics.asterix.metadata.entities.Function f = new edu.uci.ics.asterix.metadata.entities.Function(
+ dataverse, libraryName + ":" + function.getName(), args.size(), args, function.getReturnType(),
+ function.getDefinition(), library.getLanguage(), function.getFunctionType());
+ MetadataManager.INSTANCE.addFunction(mdTxnCtx, f);
+ }
+
+ MetadataManager.INSTANCE.addLibrary(mdTxnCtx, new edu.uci.ics.asterix.metadata.entities.Library(dataverse,
+ libraryName));
+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
+ } catch (Exception e) {
+ e.printStackTrace();
+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
+ }
+ }
+
+ private static void registerLibrary(String dataverse, String libraryName, boolean isMetadataNode, File installLibDir)
+ throws Exception {
+ ClassLoader classLoader = getLibraryClassLoader(dataverse, libraryName);
+ ExternalLibraryManager.registerLibraryClassLoader(dataverse, libraryName, classLoader);
+ }
+
+ private static Library getLibrary(File libraryXMLPath) throws Exception {
+ JAXBContext configCtx = JAXBContext.newInstance(Library.class);
+ Unmarshaller unmarshaller = configCtx.createUnmarshaller();
+ Library library = (Library) unmarshaller.unmarshal(libraryXMLPath);
+ return library;
+ }
+
+ private static ClassLoader getLibraryClassLoader(String dataverse, String libraryName) throws Exception {
+ System.out.println(" installing lirbary " + libraryName + " in dataverse " + dataverse);
+ File installDir = getLibraryInstallDir();
+ System.out.println(" install directory " + installDir.getAbsolutePath());
+
+ File libDir = new File(installDir.getAbsolutePath() + File.separator + dataverse + File.separator + libraryName);
+ FilenameFilter jarFileFilter = new FilenameFilter() {
+ public boolean accept(File dir, String name) {
+ return name.endsWith(".jar");
+ }
+ };
+
+ String[] jarsInLibDir = libDir.list(jarFileFilter);
+ System.out.println(" jars in lib dir " + jarsInLibDir);
+
+ if (jarsInLibDir.length > 1) {
+ throw new Exception("Incorrect library structure: found multiple library jars");
+ }
+ if (jarsInLibDir.length < 0) {
+ throw new Exception("Incorrect library structure: could not find library jar");
+ }
+
+ File libJar = new File(libDir, jarsInLibDir[0]);
+ File libDependencyDir = new File(libDir.getAbsolutePath() + File.separator + "lib");
+ int numDependencies = 1;
+ String[] libraryDependencies = null;
+ if (libDependencyDir.exists()) {
+ libraryDependencies = libDependencyDir.list(jarFileFilter);
+ numDependencies += libraryDependencies.length;
+ }
+
+ ClassLoader parentClassLoader = ExternalLibraryBootstrap.class.getClassLoader();
+ URL[] urls = new URL[numDependencies];
+ int count = 0;
+ urls[count++] = libJar.toURL();
+
+ if (libraryDependencies != null && libraryDependencies.length > 0) {
+ for (String dependency : libraryDependencies) {
+ File file = new File(libDependencyDir + File.separator + dependency);
+ urls[count++] = file.toURL();
+ }
+ }
+ ClassLoader classLoader = new URLClassLoader(urls, parentClassLoader);
+ return classLoader;
+ }
+
+ private static File getLibraryInstallDir() {
+ String workingDir = System.getProperty("user.dir");
+ return new File(workingDir + File.separator + "library");
+ }
+
+ private static File getLibraryUninstallDir() {
+ String workingDir = System.getProperty("user.dir");
+ return new File(workingDir + File.separator + "uninstall");
+ }
+
+}
+
+class ExternalLibrary {
+
+ private final String dataverse;
+ private final String name;
+ private final String language;
+ private final Functions functions;
+
+ public ExternalLibrary(String dataverse, String name, String language, Functions functions) {
+ this.dataverse = dataverse;
+ this.name = name;
+ this.language = language;
+ this.functions = functions;
+ }
+
+ public String toString() {
+ StringBuilder builder = new StringBuilder("");
+ builder.append("Library");
+ builder.append("\n");
+ builder.append("Functions");
+ builder.append("\n");
+ for (Function function : functions.getFunction()) {
+ builder.append(function);
+ builder.append("\n");
+ }
+ return new String(builder);
+ }
+
+ public String getDataverse() {
+ return dataverse;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public String getLanguage() {
+ return language;
+ }
+
+ public Functions getFunctions() {
+ return functions;
+ }
+
+}
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/NCApplicationEntryPoint.java b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/NCApplicationEntryPoint.java
index f332155..c4404cc 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/NCApplicationEntryPoint.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/hyracks/bootstrap/NCApplicationEntryPoint.java
@@ -119,6 +119,7 @@
systemState == SystemState.NEW_UNIVERSE);
MetadataBootstrap.startDDLRecovery();
}
+ ExternalLibraryBootstrap.setUpExternaLibraries(isMetadataNode);
IRecoveryManager recoveryMgr = runtimeContext.getTransactionSubsystem()
.getRecoveryManager();
@@ -167,4 +168,4 @@
}
}
-}
\ No newline at end of file
+}
diff --git a/asterix-aql/src/main/java/edu/uci/ics/asterix/aql/rewrites/AqlRewriter.java b/asterix-aql/src/main/java/edu/uci/ics/asterix/aql/rewrites/AqlRewriter.java
index 62e40fa..d295ffb 100644
--- a/asterix-aql/src/main/java/edu/uci/ics/asterix/aql/rewrites/AqlRewriter.java
+++ b/asterix-aql/src/main/java/edu/uci/ics/asterix/aql/rewrites/AqlRewriter.java
@@ -161,34 +161,41 @@
continue;
}
- FunctionDecl functionDecl = lookupUserDefinedFunctionDecl(signature);
- if (functionDecl != null) {
- if (functionDecls.contains(functionDecl)) {
- throw new AsterixException(" Detected recursvity!");
- } else {
- functionDecls.add(functionDecl);
- buildOtherUdfs(functionDecl.getFuncBody(), functionDecls, declaredFunctions);
- }
- } else {
+ Function function = lookupUserDefinedFunctionDecl(signature);
+ if (function == null) {
if (isBuiltinFunction(signature)) {
continue;
+ }
+ StringBuilder messageBuilder = new StringBuilder();
+ if (functionDecls.size() > 0) {
+ messageBuilder.append(" function " + functionDecls.get(functionDecls.size() - 1).getSignature()
+ + " depends upon function " + signature + " which is undefined");
} else {
- throw new AsterixException(" unknown function " + signature);
+ messageBuilder.append(" function " + signature + " is undefined ");
+ }
+ throw new AsterixException(messageBuilder.toString());
+ }
+
+ if (function.getLanguage().equalsIgnoreCase(Function.LANGUAGE_AQL)) {
+ FunctionDecl functionDecl = FunctionUtils.getFunctionDecl(function);
+ if (functionDecl != null) {
+ if (functionDecls.contains(functionDecl)) {
+ throw new AsterixException("ERROR:Recursive invocation "
+ + functionDecls.get(functionDecls.size() - 1).getSignature() + " <==> "
+ + functionDecl.getSignature());
+ }
+ functionDecls.add(functionDecl);
+ buildOtherUdfs(functionDecl.getFuncBody(), functionDecls, declaredFunctions);
}
}
}
}
- private FunctionDecl lookupUserDefinedFunctionDecl(FunctionSignature signature) throws AsterixException {
+ private Function lookupUserDefinedFunctionDecl(FunctionSignature signature) throws AsterixException {
if (signature.getNamespace() == null) {
return null;
}
- Function function = MetadataManager.INSTANCE.getFunction(mdTxnCtx, signature);
- if (function == null) {
- return null;
- }
- return FunctionUtils.getFunctionDecl(function);
-
+ return MetadataManager.INSTANCE.getFunction(mdTxnCtx, signature);
}
private boolean isBuiltinFunction(FunctionSignature functionSignature) {
diff --git a/asterix-aql/src/main/javacc/AQL.jj b/asterix-aql/src/main/javacc/AQL.jj
index 8e3e203..a69bbe6 100644
--- a/asterix-aql/src/main/javacc/AQL.jj
+++ b/asterix-aql/src/main/javacc/AQL.jj
@@ -2050,15 +2050,30 @@
List<Expression> argList = new ArrayList<Expression>();
Expression tmp;
int arity = 0;
- String funcName;
- String dataverse;
+ String funcName=null;
+ String dataverse=null;
+ String libraryName=null;
String hint=null;
String id1=null;
String id2=null;
+ String first = null;
+ String second = null;
+ String third = null;
+ Triple<String,String,String> functionCallComponents = null;
}
{
-
- <IDENTIFIER> { dataverse = defaultDataverse; funcName = token.image;} ("." <IDENTIFIER> { dataverse = funcName; funcName = token.image;})?
+ (
+ ( functionCallComponents = getFunctionCallComponents() )
+ {
+ dataverse = functionCallComponents.first == null ? defaultDataverse : functionCallComponents.first;
+ libraryName = functionCallComponents.second;
+ funcName = functionCallComponents.third;
+ if(libraryName != null){
+ funcName = libraryName + ":" + funcName;
+ }
+ }
+ )
+
{
hint = getHint(token);
}
@@ -2069,10 +2084,10 @@
} ("," tmp = Expression() { argList.add(tmp); arity++; })*)? <RIGHTPAREN>
{
- FunctionSignature signature = lookupFunctionSignature(dataverse, funcName.toString(), arity);
+ FunctionSignature signature = lookupFunctionSignature(dataverse, funcName, arity);
if(signature == null)
{
- signature = new FunctionSignature(dataverse, funcName.toString(), arity);
+ signature = new FunctionSignature(dataverse, funcName, arity);
}
callExpr = new CallExpr(signature,argList);
if (hint != null && hint.startsWith(INDEXED_NESTED_LOOP_JOIN_HINT)) {
@@ -2082,6 +2097,43 @@
}
}
+Triple<String,String,String> getFunctionCallComponents() throws ParseException:
+{
+ String first = null;
+ String second = null;
+ String third = null;
+}
+{
+ <IDENTIFIER>{
+ first = token.image;
+ }
+ (
+ ("." <IDENTIFIER> {
+ second = token.image;
+ }
+ )?
+ ( ":" <IDENTIFIER> {
+ third = token.image;
+ }
+ )?
+ )
+ {
+ if(second == null) {
+ if(third!=null){
+ second="" + first;
+ } else {
+ third = "" + first;
+ }
+ first = null;
+ } else if (third == null) {
+ third = "" + second;
+ second = null;
+ }
+ return new Triple<String,String,String>(first,second,third);
+ }
+
+}
+
Expression DatasetAccessExpression() throws ParseException:
{
CallExpr callExpr;
diff --git a/asterix-events/src/main/java/edu/uci/ics/asterix/event/driver/EventDriver.java b/asterix-events/src/main/java/edu/uci/ics/asterix/event/driver/EventDriver.java
index 66c0102..b8f0fc6 100644
--- a/asterix-events/src/main/java/edu/uci/ics/asterix/event/driver/EventDriver.java
+++ b/asterix-events/src/main/java/edu/uci/ics/asterix/event/driver/EventDriver.java
@@ -159,4 +159,5 @@
}
EventUtil.executeLocalScript(clientNode, eventsDir + "/" + "events" + "/" + "cleanup.sh", args);
}
+
}
diff --git a/asterix-events/src/main/java/edu/uci/ics/asterix/event/management/EventUtil.java b/asterix-events/src/main/java/edu/uci/ics/asterix/event/management/EventUtil.java
index e6c58eb..d484947 100644
--- a/asterix-events/src/main/java/edu/uci/ics/asterix/event/management/EventUtil.java
+++ b/asterix-events/src/main/java/edu/uci/ics/asterix/event/management/EventUtil.java
@@ -31,232 +31,215 @@
public class EventUtil {
- public static final String EVENTS_DIR = "events";
- public static final String CLUSTER_CONF = "config/cluster.xml";
- public static final String PATTERN_CONF = "config/pattern.xml";
- public static final DateFormat dateFormat = new SimpleDateFormat(
- "yyyy/MM/dd HH:mm:ss");
+ public static final String EVENTS_DIR = "events";
+ public static final String CLUSTER_CONF = "config/cluster.xml";
+ public static final String PATTERN_CONF = "config/pattern.xml";
+ public static final DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
- private static final String IP_LOCATION = "IP_LOCATION";
- private static final String CLUSTER_ENV = "ENV";
- private static final String SCRIPT = "SCRIPT";
- private static final String ARGS = "ARGS";
- private static final String EXECUTE_SCRIPT = "events/execute.sh";
+ private static final String IP_LOCATION = "IP_LOCATION";
+ private static final String CLUSTER_ENV = "ENV";
+ private static final String SCRIPT = "SCRIPT";
+ private static final String ARGS = "ARGS";
+ private static final String EXECUTE_SCRIPT = "events/execute.sh";
- public static long parseTimeInterval(ValueType v, String unit)
- throws IllegalArgumentException {
- int val = 0;
- switch (v.getType()) {
- case ABS:
- val = Integer.parseInt(v.getAbsoluteValue());
- break;
- case RANDOM_MIN_MAX:
- val = Randomizer.getInstance().getRandomInt(v.getMin(), v.getMax());
- break;
- case RANDOM_RANGE:
- String[] values = v.getRangeSet();
- val = Integer.parseInt(values[Randomizer.getInstance()
- .getRandomInt(0, values.length - 1)]);
- break;
- }
- return computeInterval(val, unit);
- }
+ public static long parseTimeInterval(ValueType v, String unit) throws IllegalArgumentException {
+ int val = 0;
+ switch (v.getType()) {
+ case ABS:
+ val = Integer.parseInt(v.getAbsoluteValue());
+ break;
+ case RANDOM_MIN_MAX:
+ val = Randomizer.getInstance().getRandomInt(v.getMin(), v.getMax());
+ break;
+ case RANDOM_RANGE:
+ String[] values = v.getRangeSet();
+ val = Integer.parseInt(values[Randomizer.getInstance().getRandomInt(0, values.length - 1)]);
+ break;
+ }
+ return computeInterval(val, unit);
+ }
- public static long parseTimeInterval(String v, String unit)
- throws IllegalArgumentException {
- int value = Integer.parseInt(v);
- return computeInterval(value, unit);
- }
+ public static long parseTimeInterval(String v, String unit) throws IllegalArgumentException {
+ int value = Integer.parseInt(v);
+ return computeInterval(value, unit);
+ }
- private static long computeInterval(int val, String unit) {
- int vmult = 1;
- if ("hr".equalsIgnoreCase(unit)) {
- vmult = 3600 * 1000;
- } else if ("min".equalsIgnoreCase(unit)) {
- vmult = 60 * 1000;
- } else if ("sec".equalsIgnoreCase(unit)) {
- vmult = 1000;
- } else
- throw new IllegalArgumentException(
- " invalid unit value specified for frequency (hr,min,sec)");
- return val * vmult;
+ private static long computeInterval(int val, String unit) {
+ int vmult = 1;
+ if ("hr".equalsIgnoreCase(unit)) {
+ vmult = 3600 * 1000;
+ } else if ("min".equalsIgnoreCase(unit)) {
+ vmult = 60 * 1000;
+ } else if ("sec".equalsIgnoreCase(unit)) {
+ vmult = 1000;
+ } else
+ throw new IllegalArgumentException(" invalid unit value specified for frequency (hr,min,sec)");
+ return val * vmult;
- }
+ }
- public static Event getEvent(Pattern pattern, Events events) {
- for (Event event : events.getEvent()) {
- if (event.getType().equals(pattern.getEvent().getType())) {
- return event;
- }
- }
- throw new IllegalArgumentException(" Unknown event type"
- + pattern.getEvent().getType());
- }
+ public static Event getEvent(Pattern pattern, Events events) {
+ for (Event event : events.getEvent()) {
+ if (event.getType().equals(pattern.getEvent().getType())) {
+ return event;
+ }
+ }
+ throw new IllegalArgumentException(" Unknown event type" + pattern.getEvent().getType());
+ }
- public static Node getEventLocation(Pattern pattern,
- List<Node> candidateLocations, Cluster cluster) {
- ValueType value = new ValueType(pattern.getEvent().getNodeid()
- .getValue());
- Node location = null;
- Type vtype = value.getType();
+ public static Node getEventLocation(Pattern pattern, List<Node> candidateLocations, Cluster cluster) {
+ ValueType value = new ValueType(pattern.getEvent().getNodeid().getValue());
+ Node location = null;
+ Type vtype = value.getType();
- switch (vtype) {
- case ABS:
- location = getNodeFromId(value.getAbsoluteValue(), cluster);
- break;
- case RANDOM_RANGE:
- int nodeIndex = Randomizer.getInstance().getRandomInt(0,
- candidateLocations.size() - 1);
- location = candidateLocations.get(nodeIndex);
- break;
- case RANDOM_MIN_MAX:
- throw new IllegalStateException(
- " Canont configure a min max value range for location");
- }
- return location;
+ switch (vtype) {
+ case ABS:
+ location = getNodeFromId(value.getAbsoluteValue(), cluster);
+ break;
+ case RANDOM_RANGE:
+ int nodeIndex = Randomizer.getInstance().getRandomInt(0, candidateLocations.size() - 1);
+ location = candidateLocations.get(nodeIndex);
+ break;
+ case RANDOM_MIN_MAX:
+ throw new IllegalStateException(" Canont configure a min max value range for location");
+ }
+ return location;
- }
+ }
- public static List<Node> getCandidateLocations(Pattern pattern,
- Cluster cluster) {
- ValueType value = new ValueType(pattern.getEvent().getNodeid()
- .getValue());
- List<Node> candidateList = new ArrayList<Node>();
- switch (value.getType()) {
- case ABS:
- candidateList.add(getNodeFromId(value.getAbsoluteValue(), cluster));
- break;
- case RANDOM_RANGE:
- boolean anyOption = false;
- String[] values = value.getRangeSet();
- for (String v : values) {
- if (v.equalsIgnoreCase("ANY")) {
- anyOption = true;
- }
- }
- if (anyOption) {
- for (Node node : cluster.getNode()) {
- candidateList.add(node);
- }
- } else {
- boolean found = false;
- for (String v : values) {
- for (Node node : cluster.getNode()) {
- if (node.getId().equals(v)) {
- candidateList.add(node);
- found = true;
- break;
- }
- }
- if (!found) {
- throw new IllegalStateException("Unknonw nodeId : " + v);
- }
- found = false;
- }
+ public static List<Node> getCandidateLocations(Pattern pattern, Cluster cluster) {
+ ValueType value = new ValueType(pattern.getEvent().getNodeid().getValue());
+ List<Node> candidateList = new ArrayList<Node>();
+ switch (value.getType()) {
+ case ABS:
+ candidateList.add(getNodeFromId(value.getAbsoluteValue(), cluster));
+ break;
+ case RANDOM_RANGE:
+ boolean anyOption = false;
+ String[] values = value.getRangeSet();
+ for (String v : values) {
+ if (v.equalsIgnoreCase("ANY")) {
+ anyOption = true;
+ }
+ }
+ if (anyOption) {
+ for (Node node : cluster.getNode()) {
+ candidateList.add(node);
+ }
+ } else {
+ boolean found = false;
+ for (String v : values) {
+ for (Node node : cluster.getNode()) {
+ if (node.getId().equals(v)) {
+ candidateList.add(node);
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ throw new IllegalStateException("Unknonw nodeId : " + v);
+ }
+ found = false;
+ }
- }
- String[] excluded = value.getRangeExcluded();
- if (excluded != null && excluded.length > 0) {
- List<Node> markedForRemoval = new ArrayList<Node>();
- for (String exclusion : excluded) {
- for (Node node : candidateList) {
- if (node.getId().equals(exclusion)) {
- markedForRemoval.add(node);
- }
- }
- }
- candidateList.removeAll(markedForRemoval);
- }
- break;
- case RANDOM_MIN_MAX:
- throw new IllegalStateException(
- " Invalid value configured for location");
- }
- return candidateList;
- }
+ }
+ String[] excluded = value.getRangeExcluded();
+ if (excluded != null && excluded.length > 0) {
+ List<Node> markedForRemoval = new ArrayList<Node>();
+ for (String exclusion : excluded) {
+ for (Node node : candidateList) {
+ if (node.getId().equals(exclusion)) {
+ markedForRemoval.add(node);
+ }
+ }
+ }
+ candidateList.removeAll(markedForRemoval);
+ }
+ break;
+ case RANDOM_MIN_MAX:
+ throw new IllegalStateException(" Invalid value configured for location");
+ }
+ return candidateList;
+ }
- private static Node getNodeFromId(String nodeid, Cluster cluster) {
- if (nodeid.equals(EventDriver.CLIENT_NODE.getId())) {
- return EventDriver.CLIENT_NODE;
- }
+ private static Node getNodeFromId(String nodeid, Cluster cluster) {
+ if (nodeid.equals(EventDriver.CLIENT_NODE.getId())) {
+ return EventDriver.CLIENT_NODE;
+ }
- if (nodeid.equals(cluster.getMasterNode().getId())) {
- String javaOpts = cluster.getMasterNode().getJavaOpts() == null ? cluster
- .getJavaOpts() : cluster.getMasterNode().getJavaOpts();
- String logDir = cluster.getMasterNode().getLogdir() == null ? cluster
- .getLogdir() : cluster.getMasterNode().getLogdir();
- String javaHome = cluster.getMasterNode().getJavaHome() == null ? cluster
- .getJavaHome() : cluster.getMasterNode().getJavaHome();
- BigInteger debug = cluster.getMasterNode().getDebug();
- return new Node(cluster.getMasterNode().getId(), cluster
- .getMasterNode().getClusterIp(), javaHome, javaOpts,
- logDir, null, null, debug);
- }
+ if (nodeid.equals(cluster.getMasterNode().getId())) {
+ String javaOpts = cluster.getMasterNode().getJavaOpts() == null ? cluster.getJavaOpts() : cluster
+ .getMasterNode().getJavaOpts();
+ String logDir = cluster.getMasterNode().getLogdir() == null ? cluster.getLogdir() : cluster.getMasterNode()
+ .getLogdir();
+ String javaHome = cluster.getMasterNode().getJavaHome() == null ? cluster.getJavaHome() : cluster
+ .getMasterNode().getJavaHome();
+ BigInteger debug = cluster.getMasterNode().getDebug();
+ return new Node(cluster.getMasterNode().getId(), cluster.getMasterNode().getClusterIp(), javaHome,
+ javaOpts, logDir, null, null, debug);
+ }
- List<Node> nodeList = cluster.getNode();
- for (Node node : nodeList) {
- if (node.getId().equals(nodeid)) {
- return node;
- }
- }
- StringBuffer buffer = new StringBuffer();
- buffer.append(EventDriver.CLIENT_NODE.getId() + ",");
- buffer.append(cluster.getMasterNode().getId() + ",");
- for (Node v : cluster.getNode()) {
- buffer.append(v.getId() + ",");
- }
- buffer.deleteCharAt(buffer.length() - 1);
- throw new IllegalArgumentException("Unknown node id :" + nodeid
- + " valid ids:" + buffer);
- }
+ List<Node> nodeList = cluster.getNode();
+ for (Node node : nodeList) {
+ if (node.getId().equals(nodeid)) {
+ return node;
+ }
+ }
+ StringBuffer buffer = new StringBuffer();
+ buffer.append(EventDriver.CLIENT_NODE.getId() + ",");
+ buffer.append(cluster.getMasterNode().getId() + ",");
+ for (Node v : cluster.getNode()) {
+ buffer.append(v.getId() + ",");
+ }
+ buffer.deleteCharAt(buffer.length() - 1);
+ throw new IllegalArgumentException("Unknown node id :" + nodeid + " valid ids:" + buffer);
+ }
- public static void executeEventScript(Node node, String script,
- List<String> args, Cluster cluster) throws IOException,
- InterruptedException {
- List<String> pargs = new ArrayList<String>();
- pargs.add("/bin/bash");
- pargs.add(EventDriver.getEventsDir() + "/" + EXECUTE_SCRIPT);
- StringBuffer argBuffer = new StringBuffer();
- String env = EventDriver.getStringifiedEnv(cluster) + " " + IP_LOCATION
- + "=" + node.getClusterIp();
- if (args != null) {
- for (String arg : args) {
- argBuffer.append(arg + " ");
- }
- }
- ProcessBuilder pb = new ProcessBuilder(pargs);
- pb.environment().putAll(EventDriver.getEnvironment());
- pb.environment().put(IP_LOCATION, node.getClusterIp());
- pb.environment().put(CLUSTER_ENV, env);
- pb.environment().put(SCRIPT, script);
- pb.environment().put(ARGS, argBuffer.toString());
- pb.start();
- }
+ public static void executeEventScript(Node node, String script, List<String> args, Cluster cluster)
+ throws IOException, InterruptedException {
+ List<String> pargs = new ArrayList<String>();
+ pargs.add("/bin/bash");
+ pargs.add(EventDriver.getEventsDir() + "/" + EXECUTE_SCRIPT);
+ StringBuffer argBuffer = new StringBuffer();
+ String env = EventDriver.getStringifiedEnv(cluster) + " " + IP_LOCATION + "=" + node.getClusterIp();
+ if (args != null) {
+ for (String arg : args) {
+ argBuffer.append(arg + " ");
+ }
+ }
+ ProcessBuilder pb = new ProcessBuilder(pargs);
+ pb.environment().putAll(EventDriver.getEnvironment());
+ pb.environment().put(IP_LOCATION, node.getClusterIp());
+ pb.environment().put(CLUSTER_ENV, env);
+ pb.environment().put(SCRIPT, script);
+ pb.environment().put(ARGS, argBuffer.toString());
+ pb.start();
+ }
- public static void executeLocalScript(Node node, String script,
- List<String> args) throws IOException, InterruptedException {
- List<String> pargs = new ArrayList<String>();
- pargs.add("/bin/bash");
- pargs.add(script);
- if (args != null) {
- pargs.addAll(args);
- }
- ProcessBuilder pb = new ProcessBuilder(pargs);
- pb.environment().putAll(EventDriver.getEnvironment());
- pb.environment().put(IP_LOCATION, node.getClusterIp());
- pb.start();
- }
+ public static void executeLocalScript(Node node, String script, List<String> args) throws IOException,
+ InterruptedException {
+ List<String> pargs = new ArrayList<String>();
+ pargs.add("/bin/bash");
+ pargs.add(script);
+ if (args != null) {
+ pargs.addAll(args);
+ }
+ ProcessBuilder pb = new ProcessBuilder(pargs);
+ pb.environment().putAll(EventDriver.getEnvironment());
+ pb.environment().put(IP_LOCATION, node.getClusterIp());
+ pb.start();
+ }
- public static List<String> getEventArgs(Pattern pattern) {
- List<String> pargs = new ArrayList<String>();
- if (pattern.getEvent().getPargs() == null) {
- return pargs;
- }
- String[] args = pattern.getEvent().getPargs().split(" ");
- for (String arg : args) {
- pargs.add(arg.trim());
- }
- return pargs;
- }
+ public static List<String> getEventArgs(Pattern pattern) {
+ List<String> pargs = new ArrayList<String>();
+ if (pattern.getEvent().getPargs() == null) {
+ return pargs;
+ }
+ String[] args = pattern.getEvent().getPargs().split(" ");
+ for (String arg : args) {
+ pargs.add(arg.trim());
+ }
+ return pargs;
+ }
}
diff --git a/asterix-events/src/main/resources/events/backup/backup.sh b/asterix-events/src/main/resources/events/backup/backup.sh
index 556ca39..5665cbe 100755
--- a/asterix-events/src/main/resources/events/backup/backup.sh
+++ b/asterix-events/src/main/resources/events/backup/backup.sh
@@ -1,5 +1,6 @@
WORKING_DIR=$1
ASTERIX_INSTANCE_NAME=$2
+<<<<<<< .working
ASTERIX_IODEVICES=$3
NODE_STORE=$4
ASTERIX_ROOT_METADATA_DIR=$5
@@ -8,9 +9,21 @@
BACKUP_DIR=$8
BACKUP_TYPE=$9
NODE_ID=${10}
+=======
+ASTERIX_DATA_DIR=$3
+BACKUP_ID=$4
+BACKUP_DIR=$5
+BACKUP_TYPE=$6
+NODE_ID=$7
+>>>>>>> .merge-right.r1677
+<<<<<<< .working
nodeIODevices=$(echo $ASTERIX_IODEVICES | tr "," "\n")
+=======
+nodeStores=$(echo $ASTERIX_DATA_DIR | tr "," "\n")
+>>>>>>> .merge-right.r1677
+<<<<<<< .working
if [ $BACKUP_TYPE == "hdfs" ];
then
HDFS_URL=${11}
@@ -70,3 +83,30 @@
index=`expr $index + 1`
done
fi
+=======
+if [ $BACKUP_TYPE == "hdfs" ];
+then
+ HDFS_URL=$8
+ HADOOP_VERSION=$9
+ export HADOOP_HOME=$WORKING_DIR/hadoop-$HADOOP_VERSION
+ for nodeStore in $nodeStores
+ do
+ MANGLED_DIR_NAME=`echo $nodeStores | tr / _`
+ NODE_BACKUP_DIR=$BACKUP_DIR/$ASTERIX_INSTANCE_NAME/$BACKUP_ID/$NODE_ID/$MANGLED_DIR_NAME
+ echo "$HADOOP_HOME/bin/hadoop fs -copyFromLocal $nodeStore/$NODE_ID/$ASTERIX_INSTANCE_NAME/ $HDFS_URL/$NODE_BACKUP_DIR/" >> ~/backup.log
+ $HADOOP_HOME/bin/hadoop fs -copyFromLocal $nodeStore/$NODE_ID/$ASTERIX_INSTANCE_NAME/ $HDFS_URL/$NODE_BACKUP_DIR/
+ done
+else
+ for nodeStore in $nodeStores
+ do
+ MANGLED_DIR_NAME=`echo $nodeStores | tr / _`
+ NODE_BACKUP_DIR=$BACKUP_DIR/$ASTERIX_INSTANCE_NAME/$BACKUP_ID/$NODE_ID/$MANGLED_DIR_NAME
+ if [ ! -d $NODE_BACKUP_DIR ];
+ then
+ mkdir -p $NODE_BACKUP_DIR
+ fi
+ echo "cp -r $nodeStore/$NODE_ID/$ASTERIX_INSTANCE_NAME/* $NODE_BACKUP_DIR/" >> ~/backup.log
+ cp -r $nodeStore/$NODE_ID/$ASTERIX_INSTANCE_NAME/* $NODE_BACKUP_DIR/
+ done
+fi
+>>>>>>> .merge-right.r1677
diff --git a/asterix-events/src/main/resources/events/cc_start/cc_start.sh b/asterix-events/src/main/resources/events/cc_start/cc_start.sh
index 37645bc..003d9cf 100755
--- a/asterix-events/src/main/resources/events/cc_start/cc_start.sh
+++ b/asterix-events/src/main/resources/events/cc_start/cc_start.sh
@@ -2,4 +2,5 @@
then
mkdir -p $LOG_DIR
fi
+cd $WORKING_DIR
$ASTERIX_HOME/bin/asterixcc -client-net-ip-address $CLIENT_NET_IP -client-net-port 1098 -cluster-net-ip-address $CLUSTER_NET_IP -cluster-net-port 1099 -http-port 8888 &> $LOG_DIR/cc.log
diff --git a/asterix-events/src/main/resources/events/events.xml b/asterix-events/src/main/resources/events/events.xml
index f85e3ea..392afe9 100644
--- a/asterix-events/src/main/resources/events/events.xml
+++ b/asterix-events/src/main/resources/events/events.xml
@@ -73,7 +73,14 @@
<type>file_delete</type>
<script>file/delete.sh</script>
<description>Deletes a file on the local file system to a remote node</description>
- <args>local_source_path destination_node destination_path</args>
+ <args>destination_node destination_path</args>
+ <daemon>false</daemon>
+ </event>
+ <event>
+ <type>file_create</type>
+ <script>file/create_file.sh</script>
+ <description>Creates a file on the local file system to a remote node</description>
+ <args>destination_node destination_path</args>
<daemon>false</daemon>
</event>
<event>
diff --git a/asterix-events/src/main/resources/events/file/create_file.sh b/asterix-events/src/main/resources/events/file/create_file.sh
new file mode 100644
index 0000000..762a2d3
--- /dev/null
+++ b/asterix-events/src/main/resources/events/file/create_file.sh
@@ -0,0 +1,3 @@
+mkdir -p $1
+echo "touch $1/$2" >> ~/file_create.log
+touch $1/$2
diff --git a/asterix-events/src/main/resources/events/node_join/nc_join.sh b/asterix-events/src/main/resources/events/node_join/nc_join.sh
index 6b5280b..d89c32f 100755
--- a/asterix-events/src/main/resources/events/node_join/nc_join.sh
+++ b/asterix-events/src/main/resources/events/node_join/nc_join.sh
@@ -5,4 +5,7 @@
then
mkdir -p $LOG_DIR
fi
+
+cd $WORKING_DIR
$ASTERIX_HOME/bin/asterixnc -node-id $NC_ID -cc-host $CC_HOST -cc-port 1099 -cluster-net-ip-address $IP_LOCATION -data-ip-address $IP_LOCATION -iodevices $IO_DEVICES -result-ip-address $IP_LOCATION &> $LOG_DIR/${NC_ID}.log
+
diff --git a/asterix-external-data/pom.xml b/asterix-external-data/pom.xml
index 67ca3df..5894ce9 100644
--- a/asterix-external-data/pom.xml
+++ b/asterix-external-data/pom.xml
@@ -64,6 +64,26 @@
</includes>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.jvnet.jaxb2.maven2</groupId>
+ <artifactId>maven-jaxb2-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>configuration</id>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ <configuration>
+ <schemaDirectory>src/main/resources/schema</schemaDirectory>
+ <schemaIncludes>
+ <include>library.xsd</include>
+ </schemaIncludes>
+ <generatePackage>edu.uci.ics.asterix.external.library</generatePackage>
+ <generateDirectory>${project.build.directory}/generated-sources/configuration</generateDirectory>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
@@ -95,6 +115,10 @@
<type>jar</type>
<scope>compile</scope>
</dependency>
+ <dependency>
+ <groupId>edu.uci.ics.hyracks</groupId>
+ <artifactId>algebricks-compiler</artifactId>
+ </dependency>
<dependency>
<groupId>com.kenai.nbpwr</groupId>
<artifactId>org-apache-commons-io</artifactId>
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/NCFileSystemAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/NCFileSystemAdapter.java
index ef39d45..7d378a8 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/NCFileSystemAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/NCFileSystemAdapter.java
@@ -36,30 +36,30 @@
*/
public class NCFileSystemAdapter extends FileSystemBasedAdapter {
- private static final long serialVersionUID = 1L;
- private FileSplit[] fileSplits;
+ private static final long serialVersionUID = 1L;
+ private FileSplit[] fileSplits;
- public NCFileSystemAdapter(IAType atype) {
- super(atype);
- }
+ public NCFileSystemAdapter(IAType atype) {
+ super(atype);
+ }
- @Override
- public void configure(Map<String, String> arguments) throws Exception {
- this.configuration = arguments;
- String[] splits = arguments.get(KEY_PATH).split(",");
- configureFileSplits(splits);
- configureFormat();
- }
+ @Override
+ public void configure(Map<String, String> arguments) throws Exception {
+ this.configuration = arguments;
+ String[] splits = arguments.get(KEY_PATH).split(",");
+ configureFileSplits(splits);
+ configureFormat();
+ }
- @Override
- public void initialize(IHyracksTaskContext ctx) throws Exception {
- this.ctx = ctx;
- }
+ @Override
+ public void initialize(IHyracksTaskContext ctx) throws Exception {
+ this.ctx = ctx;
+ }
- @Override
- public AdapterType getAdapterType() {
- return AdapterType.READ;
- }
+ @Override
+ public AdapterType getAdapterType() {
+ return AdapterType.READ;
+ }
private void configureFileSplits(String[] splits) throws AsterixException {
if (fileSplits == null) {
@@ -92,24 +92,25 @@
partitionConstraint = new AlgebricksAbsolutePartitionConstraint(locs);
}
- @Override
- public InputStream getInputStream(int partition) throws IOException {
- FileSplit split = fileSplits[partition];
- File inputFile = split.getLocalFile().getFile();
- InputStream in;
- try {
- in = new FileInputStream(inputFile);
- return in;
- } catch (FileNotFoundException e) {
- throw new IOException(e);
- }
- }
+ @Override
+ public InputStream getInputStream(int partition) throws IOException {
+ FileSplit split = fileSplits[partition];
+ File inputFile = split.getLocalFile().getFile();
+ InputStream in;
+ try {
+ in = new FileInputStream(inputFile);
+ return in;
+ } catch (FileNotFoundException e) {
+ throw new IOException(e);
+ }
+ }
- @Override
- public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
- if (partitionConstraint == null) {
- configurePartitionConstraint();
- }
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint()
+ throws Exception {
+ if (partitionConstraint == null) {
+ configurePartitionConstraint();
+ }
return partitionConstraint;
}
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/AsterixExternalScalarFunctionInfo.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/AsterixExternalScalarFunctionInfo.java
new file mode 100644
index 0000000..f133d91
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/AsterixExternalScalarFunctionInfo.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2009-2012 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.external.library;
+
+import java.util.List;
+
+import edu.uci.ics.asterix.om.functions.AsterixExternalFunctionInfo;
+import edu.uci.ics.asterix.om.functions.AsterixFunction;
+import edu.uci.ics.asterix.om.typecomputer.base.IResultTypeComputer;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression.FunctionKind;
+
+public class AsterixExternalScalarFunctionInfo extends AsterixExternalFunctionInfo {
+
+ private static final long serialVersionUID = 1L;
+
+ public AsterixExternalScalarFunctionInfo(String namespace, AsterixFunction asterixFunction, IAType returnType,
+ String body, String language, List<IAType> argumentTypes, IResultTypeComputer rtc) {
+ super(namespace, asterixFunction, FunctionKind.SCALAR, argumentTypes, returnType, rtc, body, language);
+ }
+
+ public AsterixExternalScalarFunctionInfo() {
+ super();
+ }
+
+
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalFunction.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalFunction.java
new file mode 100755
index 0000000..0b31268
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalFunction.java
@@ -0,0 +1,75 @@
+package edu.uci.ics.asterix.external.library;
+
+import java.io.IOException;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
+import edu.uci.ics.asterix.om.functions.IExternalFunctionInfo;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyEvaluator;
+import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyEvaluatorFactory;
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.data.std.api.IDataOutputProvider;
+import edu.uci.ics.hyracks.data.std.util.ArrayBackedValueStorage;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+public abstract class ExternalFunction implements IExternalFunction {
+
+ protected final IExternalFunctionInfo finfo;
+ protected final IFunctionFactory externalFunctionFactory;
+ protected final IExternalFunction externalFunction;
+ protected final ICopyEvaluatorFactory[] evaluatorFactories;
+ protected final IDataOutputProvider out;
+ protected final ArrayBackedValueStorage inputVal = new ArrayBackedValueStorage();
+ protected final ICopyEvaluator[] argumentEvaluators;
+ protected final JavaFunctionHelper functionHelper;
+
+ public ExternalFunction(IExternalFunctionInfo finfo, ICopyEvaluatorFactory args[],
+ IDataOutputProvider outputProvider) throws AlgebricksException {
+ this.finfo = finfo;
+ this.evaluatorFactories = args;
+ this.out = outputProvider;
+ argumentEvaluators = new ICopyEvaluator[args.length];
+ for (int i = 0; i < args.length; i++) {
+ argumentEvaluators[i] = args[i].createEvaluator(inputVal);
+ }
+ functionHelper = new JavaFunctionHelper(finfo, outputProvider);
+
+ String[] fnameComponents = finfo.getFunctionIdentifier().getName().split(":");
+ String functionLibary = fnameComponents[0];
+ String dataverse = finfo.getFunctionIdentifier().getNamespace();
+ ClassLoader libraryClassLoader = ExternalLibraryManager.getLibraryClassLoader(dataverse, functionLibary);
+ String classname = finfo.getFunctionBody();
+ Class clazz;
+ try {
+ clazz = Class.forName(classname, true, libraryClassLoader);
+ externalFunctionFactory = (IFunctionFactory) clazz.newInstance();
+ externalFunction = externalFunctionFactory.getExternalFunction();
+ } catch (Exception e) {
+ throw new AlgebricksException(" Unable to load/instantiate class " + classname, e);
+ }
+ }
+
+ public static ISerializerDeserializer getSerDe(Object typeInfo) {
+ return AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(typeInfo);
+ }
+
+ public IExternalFunctionInfo getFinfo() {
+ return finfo;
+ }
+
+ public void setArguments(IFrameTupleReference tuple) throws AlgebricksException, IOException, AsterixException {
+ for (int i = 0; i < evaluatorFactories.length; i++) {
+ inputVal.reset();
+ argumentEvaluators[i].evaluate(tuple);
+ functionHelper.setArgument(i, inputVal.getByteArray());
+ }
+ }
+
+ @Override
+ public void deinitialize() {
+ externalFunction.deinitialize();
+ }
+
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalFunctionDescriptorProvider.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalFunctionDescriptorProvider.java
new file mode 100755
index 0000000..b5888a6
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalFunctionDescriptorProvider.java
@@ -0,0 +1,51 @@
+package edu.uci.ics.asterix.external.library;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.functions.IExternalFunctionInfo;
+import edu.uci.ics.asterix.om.functions.IFunctionDescriptor;
+import edu.uci.ics.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import edu.uci.ics.hyracks.algebricks.core.algebra.functions.IFunctionInfo;
+import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyEvaluatorFactory;
+
+public class ExternalFunctionDescriptorProvider {
+
+ public static IFunctionDescriptor getExternalFunctionDescriptor(IExternalFunctionInfo finfo) throws AsterixException {
+ switch (finfo.getKind()) {
+ case SCALAR:
+ return new ExternalScalarFunctionDescriptor(finfo);
+ case AGGREGATE:
+ case UNNEST:
+ throw new AsterixException("Unsupported function kind :" + finfo.getKind());
+ default:
+ break;
+ }
+ return null;
+ }
+
+}
+
+class ExternalScalarFunctionDescriptor extends AbstractScalarFunctionDynamicDescriptor implements IFunctionDescriptor {
+
+ private final IFunctionInfo finfo;
+ private ICopyEvaluatorFactory evaluatorFactory;
+ private ICopyEvaluatorFactory[] args;
+
+ @Override
+ public ICopyEvaluatorFactory createEvaluatorFactory(ICopyEvaluatorFactory[] args) throws AlgebricksException {
+ evaluatorFactory = new ExternalScalarFunctionEvaluatorFactory((IExternalFunctionInfo) finfo, args);
+ return evaluatorFactory;
+ }
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return finfo.getFunctionIdentifier();
+ }
+
+ public ExternalScalarFunctionDescriptor(IFunctionInfo finfo) {
+ this.finfo = finfo;
+ }
+
+}
+
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalFunctionProvider.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalFunctionProvider.java
new file mode 100755
index 0000000..d895dde
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalFunctionProvider.java
@@ -0,0 +1,61 @@
+package edu.uci.ics.asterix.external.library;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import edu.uci.ics.asterix.om.functions.IExternalFunctionInfo;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyEvaluator;
+import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyEvaluatorFactory;
+import edu.uci.ics.hyracks.data.std.api.IDataOutputProvider;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+public class ExternalFunctionProvider {
+
+ private static Map<IExternalFunctionInfo, ExternalScalarFunction> functionRepo = new HashMap<IExternalFunctionInfo, ExternalScalarFunction>();
+
+ public static IExternalFunction getExternalFunctionEvaluator(IExternalFunctionInfo finfo,
+ ICopyEvaluatorFactory args[], IDataOutputProvider outputProvider) throws AlgebricksException {
+ switch (finfo.getKind()) {
+ case SCALAR:
+ ExternalScalarFunction function = functionRepo.get(finfo);
+ function = new ExternalScalarFunction(finfo, args, outputProvider);
+ // functionRepo.put(finfo, function);
+ return function;
+ case AGGREGATE:
+ case UNNEST:
+ throw new IllegalArgumentException(" not supported function kind" + finfo.getKind());
+ default:
+ throw new IllegalArgumentException(" unknown function kind" + finfo.getKind());
+ }
+ }
+}
+
+class ExternalScalarFunction extends ExternalFunction implements IExternalScalarFunction, ICopyEvaluator {
+
+ public ExternalScalarFunction(IExternalFunctionInfo finfo, ICopyEvaluatorFactory args[],
+ IDataOutputProvider outputProvider) throws AlgebricksException {
+ super(finfo, args, outputProvider);
+ initialize(functionHelper);
+ }
+
+ @Override
+ public void evaluate(IFrameTupleReference tuple) throws AlgebricksException {
+ try {
+ setArguments(tuple);
+ evaluate(functionHelper);
+ } catch (Exception e) {
+ throw new AlgebricksException(e);
+ }
+ }
+
+ public void evaluate(IFunctionHelper argumentProvider) throws Exception {
+ ((IExternalScalarFunction) externalFunction).evaluate(argumentProvider);
+ }
+
+ @Override
+ public void initialize(IFunctionHelper functionHelper) {
+ ((IExternalScalarFunction) externalFunction).initialize(functionHelper);
+ }
+
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalLibraryManager.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalLibraryManager.java
new file mode 100755
index 0000000..520020d
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalLibraryManager.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2009-2012 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.external.library;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class ExternalLibraryManager {
+
+ private static Map<String, ClassLoader> libraryClassLoaders = new HashMap<String, ClassLoader>();
+
+ public static void registerLibraryClassLoader(String dataverseName, String libraryName, ClassLoader classLoader) {
+ String key = dataverseName + "." + libraryName;
+ synchronized (libraryClassLoaders) {
+ if (libraryClassLoaders.get(dataverseName) != null) {
+ throw new IllegalStateException("library class loader already registered!");
+ }
+ libraryClassLoaders.put(key, classLoader);
+ }
+ }
+
+ public static void deregisterLibraryClassLoader(String dataverseName, String libraryName) {
+ String key = dataverseName + "." + libraryName;
+ synchronized (libraryClassLoaders) {
+ if (libraryClassLoaders.get(dataverseName) != null) {
+ libraryClassLoaders.remove(key);
+ }
+ }
+ }
+
+ public static ClassLoader getLibraryClassLoader(String dataverseName, String libraryName) {
+ String key = dataverseName + "." + libraryName;
+ synchronized (libraryClassLoaders) {
+ return libraryClassLoaders.get(key);
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalScalarFunctionEvaluatorFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalScalarFunctionEvaluatorFactory.java
new file mode 100755
index 0000000..a185000
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ExternalScalarFunctionEvaluatorFactory.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2009-2012 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.external.library;
+
+import edu.uci.ics.asterix.om.functions.IExternalFunctionInfo;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyEvaluator;
+import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyEvaluatorFactory;
+import edu.uci.ics.hyracks.data.std.api.IDataOutputProvider;
+
+public class ExternalScalarFunctionEvaluatorFactory implements ICopyEvaluatorFactory {
+
+ private final IExternalFunctionInfo finfo;
+ private final ICopyEvaluatorFactory[] args;
+
+ public ExternalScalarFunctionEvaluatorFactory(IExternalFunctionInfo finfo, ICopyEvaluatorFactory[] args)
+ throws AlgebricksException {
+ this.finfo = finfo;
+ this.args = args;
+ }
+
+ @Override
+ public ICopyEvaluator createEvaluator(IDataOutputProvider output) throws AlgebricksException {
+ return (ExternalScalarFunction) ExternalFunctionProvider.getExternalFunctionEvaluator(finfo, args, output);
+ }
+
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IExternalFunction.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IExternalFunction.java
new file mode 100755
index 0000000..e667828
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IExternalFunction.java
@@ -0,0 +1,7 @@
+package edu.uci.ics.asterix.external.library;
+
+public interface IExternalFunction {
+
+ public void deinitialize();
+
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IExternalScalarFunction.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IExternalScalarFunction.java
new file mode 100755
index 0000000..ddab3d8
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IExternalScalarFunction.java
@@ -0,0 +1,10 @@
+package edu.uci.ics.asterix.external.library;
+
+
+public interface IExternalScalarFunction extends IExternalFunction {
+
+ public void initialize(IFunctionHelper functionHelper);
+
+ public void evaluate(IFunctionHelper functionHelper) throws Exception;
+
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IFunctionFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IFunctionFactory.java
new file mode 100755
index 0000000..f67957d
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IFunctionFactory.java
@@ -0,0 +1,7 @@
+package edu.uci.ics.asterix.external.library;
+
+public interface IFunctionFactory {
+
+ public IExternalFunction getExternalFunction();
+
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IFunctionHelper.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IFunctionHelper.java
new file mode 100755
index 0000000..0192e33
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IFunctionHelper.java
@@ -0,0 +1,32 @@
+/*
+ * Copyright 2009-2012 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.external.library;
+
+import java.io.IOException;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.external.library.java.IJObject;
+import edu.uci.ics.asterix.external.library.java.JTypeTag;
+
+public interface IFunctionHelper {
+
+ public IJObject getArgument(int index);
+
+ public IJObject getResultObject();
+
+ public void setResult(IJObject result) throws IOException, AsterixException;
+
+ public IJObject getObject(JTypeTag jtypeTag);
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IResultCollector.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IResultCollector.java
new file mode 100755
index 0000000..fee002e
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/IResultCollector.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2009-2012 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.external.library;
+
+import java.io.DataOutput;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.base.AOrderedList;
+import edu.uci.ics.asterix.om.base.ARecord;
+import edu.uci.ics.asterix.om.base.IAObject;
+
+public interface IResultCollector {
+
+ public void writeIntResult(int result) throws AsterixException;
+
+ public void writeFloatResult(float result) throws AsterixException;
+
+ public void writeDoubleResult(double result) throws AsterixException;
+
+ public void writeStringResult(String result) throws AsterixException;
+
+ public void writeRecordResult(ARecord result) throws AsterixException;
+
+ public void writeListResult(AOrderedList list) throws AsterixException;
+
+ public IAObject getComplexTypeResultHolder();
+
+ public DataOutput getDataOutput();
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/JTypeObjectFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/JTypeObjectFactory.java
new file mode 100644
index 0000000..ae30695
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/JTypeObjectFactory.java
@@ -0,0 +1,37 @@
+package edu.uci.ics.asterix.external.library;
+
+import edu.uci.ics.asterix.external.library.java.IJObject;
+import edu.uci.ics.asterix.external.library.java.JObjects.JInt;
+import edu.uci.ics.asterix.external.library.java.JObjects.JRecord;
+import edu.uci.ics.asterix.external.library.java.JObjects.JString;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.container.IObjectFactory;
+
+public class JTypeObjectFactory implements IObjectFactory<IJObject, IAType> {
+
+ @Override
+ public IJObject create(IAType type) {
+ IJObject retValue = null;
+ switch (type.getTypeTag()) {
+ case INT32:
+ retValue = new JInt(0);
+ break;
+ case STRING:
+ retValue = new JString("");
+ break;
+ case RECORD:
+ IAType[] fieldTypes = ((ARecordType) type).getFieldTypes();
+ IJObject[] fieldObjects = new IJObject[fieldTypes.length];
+ int index = 0;
+ for (IAType fieldType : fieldTypes) {
+ fieldObjects[index] = create(fieldType);
+ index++;
+ }
+ retValue = new JRecord((ARecordType) type, fieldObjects);
+
+ break;
+ }
+ return retValue;
+ }
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/JavaFunctionHelper.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/JavaFunctionHelper.java
new file mode 100644
index 0000000..637d4d4
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/JavaFunctionHelper.java
@@ -0,0 +1,195 @@
+/*
+ * Copyright 2009-2012 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.external.library;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.util.List;
+
+import edu.uci.ics.asterix.builders.RecordBuilder;
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.external.library.java.IJObject;
+import edu.uci.ics.asterix.external.library.java.JObjectUtil;
+import edu.uci.ics.asterix.external.library.java.JObjects.ByteArrayAccessibleDataInputStream;
+import edu.uci.ics.asterix.external.library.java.JObjects.ByteArrayAccessibleInputStream;
+import edu.uci.ics.asterix.external.library.java.JObjects.JRecord;
+import edu.uci.ics.asterix.external.library.java.JTypeTag;
+import edu.uci.ics.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
+import edu.uci.ics.asterix.om.base.ARecord;
+import edu.uci.ics.asterix.om.base.AString;
+import edu.uci.ics.asterix.om.base.IAObject;
+import edu.uci.ics.asterix.om.functions.IExternalFunctionInfo;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.container.IObjectPool;
+import edu.uci.ics.asterix.om.util.container.ListObjectPool;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.data.std.api.IDataOutputProvider;
+import edu.uci.ics.hyracks.data.std.util.ArrayBackedValueStorage;
+
+public class JavaFunctionHelper implements IFunctionHelper {
+
+ private final IExternalFunctionInfo finfo;
+ private final IDataOutputProvider outputProvider;
+ private IJObject[] arguments;
+ private IJObject resultHolder;
+ private IAObject innerResult;
+ private ISerializerDeserializer resultSerde;
+ private IObjectPool<IJObject, IAType> objectPool = new ListObjectPool<IJObject, IAType>(new JTypeObjectFactory());
+ byte[] buffer = new byte[32768];
+ ByteArrayAccessibleInputStream bis = new ByteArrayAccessibleInputStream(buffer, 0, buffer.length);
+ ByteArrayAccessibleDataInputStream dis = new ByteArrayAccessibleDataInputStream(bis);
+
+ public JavaFunctionHelper(IExternalFunctionInfo finfo, IDataOutputProvider outputProvider)
+ throws AlgebricksException {
+ this.finfo = finfo;
+ this.outputProvider = outputProvider;
+ List<IAType> params = finfo.getParamList();
+ arguments = new IJObject[params.size()];
+ int index = 0;
+ for (IAType param : params) {
+ this.arguments[index] = objectPool.allocate(param);
+ index++;
+ }
+ resultHolder = objectPool.allocate(finfo.getReturnType());
+ }
+
+ @Override
+ public IJObject getArgument(int index) {
+ return arguments[index];
+ }
+
+ @Override
+ public void setResult(IJObject result) throws IOException, AsterixException {
+ IAObject obj = result.getIAObject();
+ try {
+ outputProvider.getDataOutput().writeByte(obj.getType().getTypeTag().serialize());
+ } catch (IOException e) {
+ throw new HyracksDataException(e);
+ }
+
+ if (obj.getType().getTypeTag().equals(ATypeTag.RECORD)) {
+ ARecordType recType = (ARecordType) obj.getType();
+ if (recType.isOpen()) {
+ writeOpenRecord((JRecord) result);
+ } else {
+ resultSerde = AqlSerializerDeserializerProvider.INSTANCE.getNonTaggedSerializerDeserializer(recType);
+ resultSerde.serialize(obj, outputProvider.getDataOutput());
+ }
+ } else {
+ resultSerde = AqlSerializerDeserializerProvider.INSTANCE.getNonTaggedSerializerDeserializer(obj.getType());
+ resultSerde.serialize(obj, outputProvider.getDataOutput());
+ }
+ reset();
+ }
+
+ private void writeOpenRecord(JRecord jRecord) throws AsterixException, IOException {
+ ARecord aRecord = (ARecord) jRecord.getIAObject();
+ RecordBuilder recordBuilder = new RecordBuilder();
+ ARecordType recordType = aRecord.getType();
+ recordBuilder.reset(recordType);
+ ArrayBackedValueStorage fieldName = new ArrayBackedValueStorage();
+ ArrayBackedValueStorage fieldValue = new ArrayBackedValueStorage();
+ List<Boolean> openField = jRecord.getOpenField();
+
+ int fieldIndex = 0;
+ int closedFieldId = 0;
+ for (IJObject field : jRecord.getFields()) {
+ fieldValue.reset();
+ switch (field.getTypeTag()) {
+ case INT32:
+ AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AINT32).serialize(
+ field.getIAObject(), fieldValue.getDataOutput());
+ break;
+ case STRING:
+ AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ASTRING)
+ .serialize(field.getIAObject(), fieldValue.getDataOutput());
+ break;
+ }
+ if (openField.get(fieldIndex)) {
+ String fName = jRecord.getFieldNames().get(fieldIndex);
+ fieldName.reset();
+ AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ASTRING).serialize(
+ new AString(fName), fieldName.getDataOutput());
+ recordBuilder.addField(fieldName, fieldValue);
+ } else {
+ recordBuilder.addField(closedFieldId, fieldValue);
+ closedFieldId++;
+ }
+ fieldIndex++;
+ }
+
+ recordBuilder.write(outputProvider.getDataOutput(), false);
+
+ }
+
+ private void reset() {
+ for (IJObject jObject : arguments) {
+ switch (jObject.getTypeTag()) {
+ case RECORD:
+ reset((JRecord) jObject);
+ break;
+ }
+ }
+ switch (resultHolder.getTypeTag()) {
+ case RECORD:
+ reset((JRecord) resultHolder);
+ break;
+ }
+ }
+
+ private void reset(JRecord jRecord) {
+ List<IJObject> fields = ((JRecord) jRecord).getFields();
+ for (IJObject field : fields) {
+ switch (field.getTypeTag()) {
+ case RECORD:
+ reset((JRecord) field);
+ break;
+ }
+ }
+ jRecord.close();
+ }
+
+ public void setArgument(int index, byte[] argument) throws IOException, AsterixException {
+ bis.setContent(argument, 1, argument.length);
+ IAType type = finfo.getParamList().get(index);
+ arguments[index] = JObjectUtil.getJType(type.getTypeTag(), type, dis, objectPool);
+ }
+
+ @Override
+ public IJObject getResultObject() {
+ return resultHolder;
+ }
+
+ @Override
+ public IJObject getObject(JTypeTag jtypeTag) {
+ IJObject retValue = null;
+ switch (jtypeTag) {
+ case INT:
+ retValue = objectPool.allocate(BuiltinType.AINT32);
+ break;
+ case STRING:
+ retValue = objectPool.allocate(BuiltinType.ASTRING);
+ break;
+ }
+ return retValue;
+ }
+
+}
\ No newline at end of file
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ResultCollector.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ResultCollector.java
new file mode 100755
index 0000000..d53b044
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/ResultCollector.java
@@ -0,0 +1,147 @@
+/*
+ * Copyright 2009-2012 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.external.library;
+
+import java.io.DataOutput;
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
+import edu.uci.ics.asterix.om.base.AMutableDouble;
+import edu.uci.ics.asterix.om.base.AMutableFloat;
+import edu.uci.ics.asterix.om.base.AMutableInt32;
+import edu.uci.ics.asterix.om.base.AMutableOrderedList;
+import edu.uci.ics.asterix.om.base.AMutableRecord;
+import edu.uci.ics.asterix.om.base.AMutableString;
+import edu.uci.ics.asterix.om.base.AOrderedList;
+import edu.uci.ics.asterix.om.base.ARecord;
+import edu.uci.ics.asterix.om.base.IAObject;
+import edu.uci.ics.asterix.om.functions.IExternalFunctionInfo;
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.data.std.api.IDataOutputProvider;
+
+public class ResultCollector implements IResultCollector {
+
+ private IAObject reusableResultObjectHolder;
+ private ByteBuffer reusableResultBinaryHolder;
+ private IDataOutputProvider outputProvider;
+ private IExternalFunctionInfo finfo;
+
+ public ResultCollector(IExternalFunctionInfo finfo, IDataOutputProvider outputProvider) {
+ this.finfo = finfo;
+ IAType returnType = finfo.getReturnType();
+ reusableResultObjectHolder = allocateResultObjectHolder(returnType);
+ reusableResultBinaryHolder = allocateResultBinaryHolder(returnType);
+ this.outputProvider = outputProvider;
+ }
+
+ private IAObject allocateResultObjectHolder(IAType type) {
+ switch (type.getTypeTag()) {
+ case INT32:
+ return new AMutableInt32(0);
+ case FLOAT:
+ return new AMutableFloat(0f);
+ case DOUBLE:
+ return new AMutableDouble(0);
+ case STRING:
+ return new AMutableString("");
+ case ORDEREDLIST:
+ return new AMutableOrderedList((AOrderedListType) type);
+ case RECORD:
+ IAType[] fieldType = ((ARecordType) type).getFieldTypes();
+ IAObject[] fieldObjects = new IAObject[fieldType.length];
+ for (int i = 0; i < fieldType.length; i++) {
+ fieldObjects[i] = allocateResultObjectHolder(fieldType[i]);
+ }
+ return new AMutableRecord((ARecordType) type, fieldObjects);
+ }
+ return null;
+ }
+
+ private ByteBuffer allocateResultBinaryHolder(IAType type) {
+ switch (type.getTypeTag()) {
+ case INT32:
+ return ByteBuffer.allocate(4);
+ case FLOAT:
+ return ByteBuffer.allocate(4);
+ case DOUBLE:
+ return ByteBuffer.allocate(8);
+ case STRING:
+ return ByteBuffer.allocate(32 * 1024);
+ case ORDEREDLIST:
+ return ByteBuffer.allocate(32 * 1024);
+ case RECORD:
+ return ByteBuffer.allocate(32 * 1024);
+ }
+ return null;
+ }
+
+ @Override
+ public void writeDoubleResult(double result) throws AsterixException {
+ ((AMutableDouble) reusableResultObjectHolder).setValue(result);
+ serializeResult(reusableResultObjectHolder);
+ }
+
+ @Override
+ public void writeFloatResult(float result) throws AsterixException {
+ ((AMutableDouble) reusableResultObjectHolder).setValue(result);
+ serializeResult(reusableResultObjectHolder);
+ }
+
+ @Override
+ public void writeIntResult(int result) throws AsterixException {
+ ((AMutableInt32) reusableResultObjectHolder).setValue(result);
+ serializeResult(reusableResultObjectHolder);
+ }
+
+ @Override
+ public void writeStringResult(String result) throws AsterixException {
+ ((AMutableString) reusableResultObjectHolder).setValue(result);
+ serializeResult(reusableResultObjectHolder);
+
+ }
+
+ @Override
+ public void writeRecordResult(ARecord result) throws AsterixException {
+ serializeResult(result);
+ }
+
+ @Override
+ public void writeListResult(AOrderedList list) throws AsterixException {
+ serializeResult(list);
+ }
+
+ public IAObject getComplexTypeResultHolder() {
+ return reusableResultObjectHolder;
+ }
+
+ private void serializeResult(IAObject object) throws AsterixException {
+ try {
+ AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(finfo.getReturnType()).serialize(
+ reusableResultObjectHolder, outputProvider.getDataOutput());
+ } catch (HyracksDataException hde) {
+ throw new AsterixException(hde);
+ }
+ }
+
+ @Override
+ public DataOutput getDataOutput() {
+ return outputProvider.getDataOutput();
+ }
+
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/RuntimeExternalFunctionUtil.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/RuntimeExternalFunctionUtil.java
new file mode 100755
index 0000000..0c6713d
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/RuntimeExternalFunctionUtil.java
@@ -0,0 +1,100 @@
+package edu.uci.ics.asterix.external.library;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+
+import edu.uci.ics.asterix.om.base.AMutableInt32;
+import edu.uci.ics.asterix.om.base.AMutableRecord;
+import edu.uci.ics.asterix.om.base.AMutableString;
+import edu.uci.ics.asterix.om.base.IAObject;
+import edu.uci.ics.asterix.om.functions.IExternalFunctionInfo;
+import edu.uci.ics.asterix.om.functions.IFunctionDescriptor;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.NotImplementedException;
+import edu.uci.ics.hyracks.algebricks.core.algebra.functions.IFunctionInfo;
+
+public class RuntimeExternalFunctionUtil {
+
+ private static Map<String, ClassLoader> libraryClassLoaders = new HashMap<String, ClassLoader>();
+
+ public static void registerLibraryClassLoader(String dataverseName, String libraryName, ClassLoader classLoader) {
+ String key = dataverseName + "." + libraryName;
+ synchronized (libraryClassLoaders) {
+ if (libraryClassLoaders.get(dataverseName) != null) {
+ throw new IllegalStateException("library class loader already registered!");
+ }
+ libraryClassLoaders.put(key, classLoader);
+ }
+ }
+
+ public static ClassLoader getLibraryClassLoader(String dataverseName, String libraryName) {
+ String key = dataverseName + "." + libraryName;
+ synchronized (libraryClassLoaders) {
+ return libraryClassLoaders.get(key);
+ }
+ }
+
+ public static IFunctionDescriptor getFunctionDescriptor(IFunctionInfo finfo) {
+ switch (((IExternalFunctionInfo) finfo).getKind()) {
+ case SCALAR:
+ return getScalarFunctionDescriptor(finfo);
+ case AGGREGATE:
+ case UNNEST:
+ case STATEFUL:
+ throw new NotImplementedException("External " + finfo.getFunctionIdentifier().getName()
+ + " not supported");
+ }
+ return null;
+ }
+
+ private static AbstractScalarFunctionDynamicDescriptor getScalarFunctionDescriptor(IFunctionInfo finfo) {
+ return new ExternalScalarFunctionDescriptor(finfo);
+ }
+
+ public static ByteBuffer allocateArgumentBuffers(IAType type) {
+ switch (type.getTypeTag()) {
+ case INT32:
+ return ByteBuffer.allocate(4);
+ case STRING:
+ return ByteBuffer.allocate(32 * 1024);
+ default:
+ return ByteBuffer.allocate(32 * 1024);
+ }
+ }
+
+ public static IAObject allocateArgumentObjects(IAType type) {
+ switch (type.getTypeTag()) {
+ case INT32:
+ return new AMutableInt32(0);
+ case STRING:
+ return new AMutableString("");
+ default:
+ return null;
+ /*
+ ARecordType recordType = (ARecordType) type;
+ IAType[] fieldTypes = recordType.getFieldTypes();
+ IAObject[] fields = new IAObject[fieldTypes.length];
+ for (int i = 0; i < fields.length; i++) {
+ fields[i] = allocateArgumentObjects(fieldTypes[i]);
+ }
+ return new AMutableRecord((ARecordType) type, fields);
+ */
+ }
+ }
+
+ public static File getExternalLibraryDeployDir(String nodeId) {
+ String filePath = null;
+ if (nodeId != null) {
+ filePath = "edu.uci.ics.hyracks.control.nc.NodeControllerService" + "/" + nodeId + "/"
+ + "applications/asterix/expanded/external-lib/libraries";
+ } else {
+ filePath = "ClusterControllerService" + "/" + "applications/asterix/expanded/external-lib/libraries";
+
+ }
+ return new File(filePath);
+ }
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/IJObject.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/IJObject.java
new file mode 100644
index 0000000..1561c42
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/IJObject.java
@@ -0,0 +1,11 @@
+package edu.uci.ics.asterix.external.library.java;
+
+import edu.uci.ics.asterix.om.base.IAObject;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+
+public interface IJObject {
+
+ public ATypeTag getTypeTag();
+
+ public IAObject getIAObject();
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/IJType.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/IJType.java
new file mode 100644
index 0000000..fcc35eb
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/IJType.java
@@ -0,0 +1,11 @@
+package edu.uci.ics.asterix.external.library.java;
+
+import edu.uci.ics.asterix.om.base.IAObject;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+
+public interface IJType {
+
+ public ATypeTag getTypeTag();
+
+ public IAObject getIAObject();
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JObjectUtil.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JObjectUtil.java
new file mode 100644
index 0000000..56ef4fb
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JObjectUtil.java
@@ -0,0 +1,400 @@
+package edu.uci.ics.asterix.external.library.java;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.ARecordSerializerDeserializer;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AStringSerializerDeserializer;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.SerializerDeserializerUtil;
+import edu.uci.ics.asterix.external.library.java.JObjects.ByteArrayAccessibleDataInputStream;
+import edu.uci.ics.asterix.external.library.java.JObjects.JBoolean;
+import edu.uci.ics.asterix.external.library.java.JObjects.JCircle;
+import edu.uci.ics.asterix.external.library.java.JObjects.JDate;
+import edu.uci.ics.asterix.external.library.java.JObjects.JDateTime;
+import edu.uci.ics.asterix.external.library.java.JObjects.JDouble;
+import edu.uci.ics.asterix.external.library.java.JObjects.JDuration;
+import edu.uci.ics.asterix.external.library.java.JObjects.JFloat;
+import edu.uci.ics.asterix.external.library.java.JObjects.JInt;
+import edu.uci.ics.asterix.external.library.java.JObjects.JInterval;
+import edu.uci.ics.asterix.external.library.java.JObjects.JLine;
+import edu.uci.ics.asterix.external.library.java.JObjects.JOrderedList;
+import edu.uci.ics.asterix.external.library.java.JObjects.JPoint;
+import edu.uci.ics.asterix.external.library.java.JObjects.JPoint3D;
+import edu.uci.ics.asterix.external.library.java.JObjects.JPolygon;
+import edu.uci.ics.asterix.external.library.java.JObjects.JRecord;
+import edu.uci.ics.asterix.external.library.java.JObjects.JRectangle;
+import edu.uci.ics.asterix.external.library.java.JObjects.JString;
+import edu.uci.ics.asterix.external.library.java.JObjects.JTime;
+import edu.uci.ics.asterix.external.library.java.JObjects.JUnorderedList;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.AUnionType;
+import edu.uci.ics.asterix.om.types.AUnorderedListType;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.EnumDeserializer;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
+import edu.uci.ics.asterix.om.util.container.IObjectPool;
+
+public class JObjectUtil {
+
+ public static IJObject getJType(ATypeTag typeTag, IAType type, ByteArrayAccessibleDataInputStream dis,
+ IObjectPool<IJObject, IAType> objectPool) throws IOException, AsterixException {
+ IJObject jObject;
+
+ switch (typeTag) {
+
+ case INT32: {
+ int v = dis.readInt();
+ jObject = objectPool.allocate(BuiltinType.AINT32);
+ ((JInt) jObject).setValue(v);
+ break;
+ }
+
+ case FLOAT: {
+ float v = dis.readFloat();
+ jObject = objectPool.allocate(BuiltinType.AFLOAT);
+ ((JFloat) jObject).setValue(v);
+ break;
+ }
+
+ case DOUBLE: {
+ double value = dis.readDouble();
+ jObject = objectPool.allocate(BuiltinType.ADOUBLE);
+ ((JDouble) jObject).setValue(value);
+ break;
+ }
+
+ case STRING: {
+ String v = dis.readUTF();
+ jObject = objectPool.allocate(BuiltinType.ASTRING);
+ ((JString) jObject).setValue(v);
+ break;
+ }
+
+ case BOOLEAN:
+ jObject = objectPool.allocate(BuiltinType.ABOOLEAN);
+ ((JBoolean) jObject).setValue(dis.readBoolean());
+ break;
+
+ case DATE: {
+ int d = dis.readInt();
+ jObject = objectPool.allocate(BuiltinType.ADATE);
+ ((JDate) jObject).setValue(d);
+ break;
+ }
+
+ case DATETIME: {
+ jObject = objectPool.allocate(BuiltinType.ADATETIME);
+ long value = dis.readLong();
+ ((JDateTime) jObject).setValue(value);
+ break;
+ }
+
+ case DURATION: {
+ jObject = objectPool.allocate(BuiltinType.ADURATION);
+ int months = dis.readInt();
+ long msecs = dis.readLong();
+ ((JDuration) jObject).setValue(months, msecs);
+ break;
+ }
+
+ case TIME: {
+ jObject = objectPool.allocate(BuiltinType.ATIME);
+ int time = dis.readInt();
+ ((JTime) jObject).setValue(time);
+ break;
+ }
+
+ case INTERVAL: {
+ jObject = objectPool.allocate(BuiltinType.AINTERVAL);
+ long start = dis.readLong();
+ long end = dis.readLong();
+ byte intervalType = dis.readByte();
+ ((JInterval) jObject).setValue(start, end, intervalType);
+ break;
+ }
+
+ case CIRCLE: {
+ jObject = objectPool.allocate(BuiltinType.ACIRCLE);
+ double x = dis.readDouble();
+ double y = dis.readDouble();
+ double radius = dis.readDouble();
+ JPoint jpoint = (JPoint) objectPool.allocate(BuiltinType.APOINT);
+ jpoint.setValue(x, y);
+ ((JCircle) jObject).setValue(jpoint, radius);
+ break;
+ }
+
+ case POINT: {
+ jObject = objectPool.allocate(BuiltinType.APOINT);
+ double x = dis.readDouble();
+ double y = dis.readDouble();
+ ((JPoint) jObject).setValue(x, y);
+ break;
+ }
+
+ case POINT3D: {
+ jObject = objectPool.allocate(BuiltinType.APOINT3D);
+ double x = dis.readDouble();
+ double y = dis.readDouble();
+ double z = dis.readDouble();
+ ((JPoint3D) jObject).setValue(x, y, z);
+ break;
+ }
+
+ case LINE: {
+ jObject = objectPool.allocate(BuiltinType.ALINE);
+ double x1 = dis.readDouble();
+ double y1 = dis.readDouble();
+ double x2 = dis.readDouble();
+ double y2 = dis.readDouble();
+ JPoint jpoint1 = (JPoint) objectPool.allocate(BuiltinType.APOINT);
+ jpoint1.setValue(x1, y1);
+ JPoint jpoint2 = (JPoint) objectPool.allocate(BuiltinType.APOINT);
+ jpoint2.setValue(x2, y2);
+ ((JLine) jObject).setValue(jpoint1, jpoint2);
+ break;
+ }
+
+ case POLYGON: {
+ jObject = objectPool.allocate(BuiltinType.APOLYGON);
+ short numberOfPoints = dis.readShort();
+ List<JPoint> points = new ArrayList<JPoint>();
+ for (int i = 0; i < numberOfPoints; i++) {
+ JPoint p1 = (JPoint) objectPool.allocate(BuiltinType.APOINT);
+ p1.setValue(dis.readDouble(), dis.readDouble());
+ points.add(p1);
+ }
+ ((JPolygon) jObject).setValue(points);
+ break;
+ }
+
+ case RECTANGLE: {
+ jObject = objectPool.allocate(BuiltinType.ARECTANGLE);
+ double x1 = dis.readDouble();
+ double y1 = dis.readDouble();
+ double x2 = dis.readDouble();
+ double y2 = dis.readDouble();
+ JPoint jpoint1 = (JPoint) objectPool.allocate(BuiltinType.APOINT);
+ jpoint1.setValue(x1, y1);
+ JPoint jpoint2 = (JPoint) objectPool.allocate(BuiltinType.APOINT);
+ jpoint2.setValue(x2, y2);
+ ((JRectangle) jObject).setValue(jpoint1, jpoint2);
+ break;
+ }
+
+ case UNORDEREDLIST: {
+ AUnorderedListType listType = (AUnorderedListType) type;
+ IAType elementType = listType.getItemType();
+ jObject = objectPool.allocate(listType);
+
+ boolean fixedSize = false;
+ ATypeTag tag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(dis.readByte());
+ switch (tag) {
+ case STRING:
+ case RECORD:
+ case ORDEREDLIST:
+ case UNORDEREDLIST:
+ case ANY:
+ fixedSize = false;
+ break;
+ default:
+ fixedSize = true;
+ break;
+ }
+ dis.readInt(); // list size
+ int numberOfitems;
+ numberOfitems = dis.readInt();
+ if (numberOfitems > 0) {
+ if (!fixedSize) {
+ for (int i = 0; i < numberOfitems; i++)
+ dis.readInt();
+ }
+ for (int i = 0; i < numberOfitems; i++) {
+ IJObject v = (IJObject) getJType(elementType.getTypeTag(), elementType, dis, objectPool);
+ ((JUnorderedList) jObject).add(v);
+ }
+ }
+
+ break;
+ }
+ case ORDEREDLIST: {
+ AOrderedListType listType = (AOrderedListType) type;
+ IAType elementType = listType.getItemType();
+ jObject = objectPool.allocate(listType);
+ boolean fixedSize = false;
+ ATypeTag tag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(dis.readByte());
+ switch (tag) {
+ case STRING:
+ case RECORD:
+ case ORDEREDLIST:
+ case UNORDEREDLIST:
+ case ANY:
+ fixedSize = false;
+ break;
+ default:
+ fixedSize = true;
+ break;
+ }
+
+ dis.readInt(); // list size
+ int numberOfitems;
+ numberOfitems = dis.readInt();
+ if (numberOfitems > 0) {
+ if (!fixedSize) {
+ for (int i = 0; i < numberOfitems; i++)
+ dis.readInt();
+ }
+ for (int i = 0; i < numberOfitems; i++) {
+ IJObject v = (IJObject) getJType(elementType.getTypeTag(), elementType, dis, objectPool);
+ ((JOrderedList) jObject).add(v);
+ }
+ }
+
+ break;
+ }
+ case RECORD:
+ ARecordType recordType = (ARecordType) type;
+ int numberOfSchemaFields = recordType.getFieldTypes().length;
+ byte[] recordBits = dis.getInputStream().getArray();
+ boolean isExpanded = false;
+ int s = dis.getInputStream().getPosition();
+ int recordOffset = s;
+ int openPartOffset = 0;
+ int offsetArrayOffset = 0;
+ int[] fieldOffsets = new int[numberOfSchemaFields];
+ IJObject[] closedFields = new IJObject[numberOfSchemaFields];
+
+ if (recordType == null) {
+ openPartOffset = s + AInt32SerializerDeserializer.getInt(recordBits, s + 6);
+ s += 8;
+ isExpanded = true;
+ } else {
+ dis.skip(4); // reading length is not required.
+ if (recordType.isOpen()) {
+ isExpanded = dis.readBoolean();
+ if (isExpanded) {
+ openPartOffset = s + dis.readInt(); // AInt32SerializerDeserializer.getInt(recordBits, s + 6);
+ } else {
+ // do nothing s += 6;
+ }
+ } else {
+ // do nothing s += 5;
+ }
+ }
+
+ if (numberOfSchemaFields > 0) {
+ int numOfSchemaFields = dis.readInt(); //s += 4;
+ int nullBitMapOffset = 0;
+ boolean hasNullableFields = NonTaggedFormatUtil.hasNullableField(recordType);
+ if (hasNullableFields) {
+ nullBitMapOffset = dis.getInputStream().getPosition();//s
+ offsetArrayOffset = dis.getInputStream().getPosition() //s
+ + (numberOfSchemaFields % 8 == 0 ? numberOfSchemaFields / 8
+ : numberOfSchemaFields / 8 + 1);
+ } else {
+ offsetArrayOffset = dis.getInputStream().getPosition();
+ }
+ for (int i = 0; i < numberOfSchemaFields; i++) {
+ fieldOffsets[i] = dis.readInt(); // AInt32SerializerDeserializer.getInt(recordBits, offsetArrayOffset) + recordOffset;
+ // offsetArrayOffset += 4;
+ }
+ for (int fieldNumber = 0; fieldNumber < numberOfSchemaFields; fieldNumber++) {
+ if (hasNullableFields) {
+ byte b1 = recordBits[nullBitMapOffset + fieldNumber / 8];
+ int p = 1 << (7 - (fieldNumber % 8));
+ if ((b1 & p) == 0) {
+ // set null value (including type tag inside)
+ //fieldValues.add(nullReference);
+ continue;
+ }
+ }
+ IAType[] fieldTypes = recordType.getFieldTypes();
+ int fieldValueLength = 0;
+ ATypeTag fieldValueTypeTag = null;
+
+ IAType fieldType = fieldTypes[fieldNumber];
+ if (fieldTypes[fieldNumber].getTypeTag() == ATypeTag.UNION) {
+ if (NonTaggedFormatUtil.isOptionalField((AUnionType) fieldTypes[fieldNumber])) {
+ fieldType = ((AUnionType) fieldTypes[fieldNumber]).getUnionList().get(
+ NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST);
+ fieldValueTypeTag = fieldType.getTypeTag();
+ fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(recordBits,
+ fieldOffsets[fieldNumber], typeTag, false);
+ }
+ } else {
+ typeTag = fieldTypes[fieldNumber].getTypeTag();
+ }
+ closedFields[fieldNumber] = getJType(typeTag, fieldType, dis, objectPool);
+ }
+ }
+ if (isExpanded) {
+ int numberOfOpenFields = dis.readInt();
+ String[] fieldNames = new String[numberOfOpenFields];
+ IAType[] fieldTypes = new IAType[numberOfOpenFields];
+ IJObject[] openFields = new IJObject[numberOfOpenFields];
+ for (int i = 0; i < numberOfOpenFields; i++) {
+ dis.readInt();
+ dis.readInt();
+ }
+ for (int i = 0; i < numberOfOpenFields; i++) {
+ fieldNames[i] = AStringSerializerDeserializer.INSTANCE.deserialize(dis).getStringValue();
+ ATypeTag openFieldTypeTag = SerializerDeserializerUtil.deserializeTag(dis);
+ openFields[i] = getJType(openFieldTypeTag, null, dis, objectPool);
+ fieldTypes[i] = openFields[i].getIAObject().getType();
+ }
+ ARecordType openPartRecType = new ARecordType(null, fieldNames, fieldTypes, true);
+ if (numberOfSchemaFields > 0) {
+ ARecordType mergedRecordType = mergeRecordTypes(recordType, openPartRecType);
+ IJObject[] mergedFields = mergeFields(closedFields, openFields);
+ jObject = objectPool.allocate(recordType);
+ return new JRecord(mergedRecordType, mergedFields);
+ } else {
+ return new JRecord(recordType, openFields);
+ }
+ } else {
+ return new JRecord(recordType, closedFields);
+ }
+
+ default:
+ throw new IllegalStateException("Argument type: " + typeTag);
+ }
+ return jObject;
+ }
+
+ private static IJObject[] mergeFields(IJObject[] closedFields, IJObject[] openFields) {
+ IJObject[] fields = new IJObject[closedFields.length + openFields.length];
+ int i = 0;
+ for (; i < closedFields.length; i++) {
+ fields[i] = closedFields[i];
+ }
+ for (int j = 0; j < openFields.length; j++) {
+ fields[closedFields.length + j] = openFields[j];
+ }
+ return fields;
+ }
+
+ private static ARecordType mergeRecordTypes(ARecordType recType1, ARecordType recType2) throws AsterixException {
+
+ String[] fieldNames = new String[recType1.getFieldNames().length + recType2.getFieldNames().length];
+ IAType[] fieldTypes = new IAType[recType1.getFieldTypes().length + recType2.getFieldTypes().length];
+
+ int i = 0;
+ for (; i < recType1.getFieldNames().length; i++) {
+ fieldNames[i] = recType1.getFieldNames()[i];
+ fieldTypes[i] = recType1.getFieldTypes()[i];
+ }
+
+ for (int j = 0; j < recType2.getFieldNames().length; i++, j++) {
+ fieldNames[i] = recType2.getFieldNames()[j];
+ fieldTypes[i] = recType2.getFieldTypes()[j];
+ }
+ return new ARecordType(null, fieldNames, fieldTypes, true);
+ }
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JObjects.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JObjects.java
new file mode 100644
index 0000000..93931e9
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JObjects.java
@@ -0,0 +1,872 @@
+package edu.uci.ics.asterix.external.library.java;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.base.ABoolean;
+import edu.uci.ics.asterix.om.base.AMutableCircle;
+import edu.uci.ics.asterix.om.base.AMutableDate;
+import edu.uci.ics.asterix.om.base.AMutableDateTime;
+import edu.uci.ics.asterix.om.base.AMutableDouble;
+import edu.uci.ics.asterix.om.base.AMutableDuration;
+import edu.uci.ics.asterix.om.base.AMutableFloat;
+import edu.uci.ics.asterix.om.base.AMutableInt32;
+import edu.uci.ics.asterix.om.base.AMutableInt64;
+import edu.uci.ics.asterix.om.base.AMutableInterval;
+import edu.uci.ics.asterix.om.base.AMutableLine;
+import edu.uci.ics.asterix.om.base.AMutableOrderedList;
+import edu.uci.ics.asterix.om.base.AMutablePoint;
+import edu.uci.ics.asterix.om.base.AMutablePoint3D;
+import edu.uci.ics.asterix.om.base.AMutablePolygon;
+import edu.uci.ics.asterix.om.base.AMutableRecord;
+import edu.uci.ics.asterix.om.base.AMutableRectangle;
+import edu.uci.ics.asterix.om.base.AMutableString;
+import edu.uci.ics.asterix.om.base.AMutableTime;
+import edu.uci.ics.asterix.om.base.AMutableUnorderedList;
+import edu.uci.ics.asterix.om.base.APoint;
+import edu.uci.ics.asterix.om.base.IAObject;
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.AUnorderedListType;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.IAType;
+
+public class JObjects {
+
+ public static abstract class JObject implements IJObject {
+
+ protected IAObject value;
+ protected byte[] bytes;
+
+ protected JObject(IAObject value) {
+ this.value = value;
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return value.getType().getTypeTag();
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return value;
+ }
+ }
+
+ public static final class JInt implements IJObject {
+
+ private AMutableInt32 value;
+
+ public JInt(int value) {
+ this.value = new AMutableInt32(value);
+ }
+
+ public void setValue(int v) {
+ if (value == null) {
+ value = new AMutableInt32(v);
+ } else {
+ ((AMutableInt32) value).setValue(v);
+ }
+ }
+
+ public void setValue(AMutableInt32 v) {
+ value = v;
+ }
+
+ public int getValue() {
+ return ((AMutableInt32) value).getIntegerValue().intValue();
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return BuiltinType.AINT32.getTypeTag();
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return value;
+ }
+
+ }
+
+ public static final class JBoolean implements IJObject {
+
+ private boolean value;
+
+ public JBoolean(boolean value) {
+ this.value = value;
+ }
+
+ public void setValue(boolean value) {
+ this.value = value;
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.BOOLEAN;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return value ? ABoolean.TRUE : ABoolean.FALSE;
+ }
+
+ }
+
+ public static final class JLong extends JObject {
+
+ public JLong(long v) {
+ super(new AMutableInt64(v));
+ }
+
+ public void setValue(long v) {
+ ((AMutableInt64) value).setValue(v);
+ }
+
+ public long getValue() {
+ return ((AMutableInt64) value).getLongValue();
+ }
+
+ }
+
+ public static final class JDouble extends JObject {
+
+ public JDouble(double v) {
+ super(new AMutableDouble(v));
+ }
+
+ public void setValue(double v) {
+ ((AMutableDouble) value).setValue(v);
+ }
+
+ public double getValue() {
+ return ((AMutableDouble) value).getDoubleValue();
+ }
+
+ }
+
+ public static final class JString extends JObject {
+
+ public JString(String v) {
+ super(new AMutableString(v));
+ }
+
+ public void setValue(String v) {
+ ((AMutableString) value).setValue(v);
+ }
+
+ public String getValue() {
+ return ((AMutableString) value).getStringValue();
+ }
+
+ }
+
+ public static final class JFloat implements IJObject {
+
+ private AMutableFloat value;
+
+ public JFloat(float v) {
+ value = new AMutableFloat(v);
+ }
+
+ public void setValue(float v) {
+ ((AMutableFloat) value).setValue(v);
+ }
+
+ public float getValue() {
+ return ((AMutableFloat) value).getFloatValue();
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return BuiltinType.AFLOAT.getTypeTag();
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return value;
+ }
+
+ }
+
+ public static final class JPoint extends JObject {
+
+ public JPoint(double x, double y) {
+ super(new AMutablePoint(x, y));
+ }
+
+ public void setValue(double x, double y) {
+ ((AMutablePoint) value).setValue(x, y);
+ }
+
+ public double getXValue() {
+ return ((AMutablePoint) value).getX();
+ }
+
+ public double getYValue() {
+ return ((AMutablePoint) value).getY();
+ }
+
+ public IAObject getValue() {
+ return value;
+ }
+
+ @Override
+ public String toString() {
+ return value.toString();
+ }
+ }
+
+ public static final class JRectangle implements IJObject {
+
+ private AMutableRectangle rect;
+
+ public JRectangle(JPoint p1, JPoint p2) {
+ rect = new AMutableRectangle((APoint) p1.getValue(), (APoint) p2.getValue());
+ }
+
+ public void setValue(JPoint p1, JPoint p2) {
+ this.rect.setValue((APoint) p1.getValue(), (APoint) p2.getValue());
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.RECTANGLE;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return rect;
+ }
+
+ @Override
+ public String toString() {
+ return rect.toString();
+ }
+
+ }
+
+ public static final class JTime implements IJObject {
+
+ private AMutableTime time;
+
+ public JTime(int timeInMillsec) {
+ time = new AMutableTime(timeInMillsec);
+ }
+
+ public void setValue(int timeInMillsec) {
+ time.setValue(timeInMillsec);
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.TIME;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return time;
+ }
+
+ @Override
+ public String toString() {
+ return time.toString();
+ }
+
+ }
+
+ public static final class JInterval implements IJObject {
+
+ private AMutableInterval interval;
+
+ public JInterval(long intervalStart, long intervalEnd) {
+ interval = new AMutableInterval(intervalStart, intervalEnd, (byte) 0);
+ }
+
+ public void setValue(long intervalStart, long intervalEnd, byte typetag) {
+ interval.setValue(intervalStart, intervalEnd, typetag);
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.INTERVAL;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return interval;
+ }
+
+ @Override
+ public String toString() {
+ return interval.toString();
+ }
+
+ public long getIntervalStart() {
+ return interval.getIntervalStart();
+ }
+
+ public long getIntervalEnd() {
+ return interval.getIntervalEnd();
+ }
+
+ public short getIntervalType() {
+ return interval.getIntervalType();
+ }
+
+ }
+
+ public static final class JDate implements IJObject {
+
+ private AMutableDate date;
+
+ public JDate(int chrononTimeInDays) {
+ date = new AMutableDate(chrononTimeInDays);
+ }
+
+ public void setValue(int chrononTimeInDays) {
+ date.setValue(chrononTimeInDays);
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.DATE;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return date;
+ }
+
+ @Override
+ public String toString() {
+ return date.toString();
+ }
+
+ }
+
+ public static final class JDateTime implements IJObject {
+
+ private AMutableDateTime dateTime;
+
+ public JDateTime(long chrononTime) {
+ dateTime = new AMutableDateTime(chrononTime);
+ }
+
+ public void setValue(long chrononTime) {
+ dateTime.setValue(chrononTime);
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.DATETIME;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return dateTime;
+ }
+
+ @Override
+ public String toString() {
+ return dateTime.toString();
+ }
+
+ }
+
+ public static final class JDuration implements IJObject {
+
+ private AMutableDuration duration;
+
+ public JDuration(int months, long milliseconds) {
+ duration = new AMutableDuration(months, milliseconds);
+ }
+
+ public void setValue(int months, long milliseconds) {
+ duration.setValue(months, milliseconds);
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.DURATION;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return duration;
+ }
+
+ @Override
+ public String toString() {
+ return duration.toString();
+ }
+
+ }
+
+ public static final class JPolygon implements IJObject {
+
+ private AMutablePolygon polygon;
+ private List<JPoint> points;
+
+ public JPolygon(List<JPoint> points) {
+ this.points = points;
+ }
+
+ public void setValue(List<JPoint> points) {
+ this.points = points;
+ polygon = null;
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.POLYGON;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ if (polygon == null) {
+ APoint[] pts = new APoint[points.size()];
+ int index = 0;
+ for (JPoint p : points) {
+ pts[index++] = (APoint) p.getIAObject();
+ }
+ polygon = new AMutablePolygon(pts);
+ }
+ return polygon;
+ }
+
+ @Override
+ public String toString() {
+ return getIAObject().toString();
+ }
+
+ }
+
+ public static final class JCircle implements IJObject {
+
+ private AMutableCircle circle;
+
+ public JCircle(JPoint center, double radius) {
+ circle = new AMutableCircle((APoint) center.getIAObject(), radius);
+ }
+
+ public void setValue(JPoint center, double radius) {
+ circle.setValue((APoint) center.getIAObject(), radius);
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.CIRCLE;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return circle;
+ }
+
+ @Override
+ public String toString() {
+ return circle.toString();
+ }
+
+ }
+
+ public static final class JLine implements IJObject {
+
+ private AMutableLine line;
+
+ public JLine(JPoint p1, JPoint p2) {
+ line = new AMutableLine((APoint) p1.getIAObject(), (APoint) p2.getIAObject());
+ }
+
+ public void setValue(JPoint p1, JPoint p2) {
+ line.setValue((APoint) p1.getIAObject(), (APoint) p2.getIAObject());
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.LINE;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return line;
+ }
+
+ @Override
+ public String toString() {
+ return line.toString();
+ }
+
+ }
+
+ public static final class JPoint3D implements IJObject {
+
+ private AMutablePoint3D value;
+
+ public JPoint3D(double x, double y, double z) {
+ value = new AMutablePoint3D(x, y, z);
+ }
+
+ public void setValue(double x, double y, double z) {
+ value.setValue(x, y, z);
+ }
+
+ public double getXValue() {
+ return ((AMutablePoint3D) value).getX();
+ }
+
+ public double getYValue() {
+ return ((AMutablePoint3D) value).getY();
+ }
+
+ public double getZValue() {
+ return ((AMutablePoint3D) value).getZ();
+ }
+
+ public IAObject getValue() {
+ return value;
+ }
+
+ @Override
+ public String toString() {
+ return value.toString();
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.POINT3D;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return value;
+ }
+ }
+
+ public static final class JOrderedList implements IJObject {
+
+ private AOrderedListType listType;
+ private List<IJObject> jObjects;
+
+ public JOrderedList(IJObject jObject) {
+ this.listType = new AOrderedListType(jObject.getIAObject().getType(), null);
+ this.jObjects = new ArrayList<IJObject>();
+ }
+
+ public void add(IJObject jObject) {
+ jObjects.add(jObject);
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.ORDEREDLIST;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ AMutableOrderedList v = new AMutableOrderedList(listType);
+ for (IJObject jObj : jObjects) {
+ v.add(jObj.getIAObject());
+ }
+ return v;
+ }
+
+ public AOrderedListType getListType() {
+ return listType;
+ }
+
+ public void addAll(Collection<IJObject> jObjectCollection) {
+ jObjects.addAll(jObjectCollection);
+ }
+
+ public void clear() {
+ jObjects.clear();
+ }
+
+ public IJObject getElement(int index) {
+ return jObjects.get(index);
+ }
+
+ public int size() {
+ return jObjects.size();
+ }
+
+ }
+
+ public static final class JUnorderedList implements IJObject {
+
+ private AUnorderedListType listType;
+ private List<IJObject> jObjects;
+
+ public JUnorderedList(IJObject jObject) {
+ this.listType = new AUnorderedListType(jObject.getIAObject().getType(), null);
+ this.jObjects = new ArrayList<IJObject>();
+ }
+
+ public void add(IJObject jObject) {
+ jObjects.add(jObject);
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return ATypeTag.UNORDEREDLIST;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ AMutableUnorderedList v = new AMutableUnorderedList(listType);
+ for (IJObject jObj : jObjects) {
+ v.add(jObj.getIAObject());
+ }
+ return v;
+ }
+
+ public AUnorderedListType getListType() {
+ return listType;
+ }
+
+ public boolean isEmpty() {
+ return jObjects.isEmpty();
+ }
+
+ public void addAll(Collection<IJObject> jObjectCollection) {
+ jObjects.addAll(jObjectCollection);
+ }
+
+ public void clear() {
+ jObjects.clear();
+ }
+
+ public IJObject getElement(int index) {
+ return jObjects.get(index);
+ }
+
+ public int size() {
+ return jObjects.size();
+ }
+
+ }
+
+ public static final class JRecord implements IJObject {
+
+ private AMutableRecord value;
+ private ARecordType recordType;
+ private List<IJObject> fields;
+ private List<String> fieldNames;
+ private List<IAType> fieldTypes;
+ private int numFieldsAdded = 0;
+ private List<Boolean> openField;
+
+ public JRecord(ARecordType recordType) {
+ this.recordType = recordType;
+ this.fields = new ArrayList<IJObject>();
+ initFieldInfo();
+ }
+
+ public JRecord(ARecordType recordType, IJObject[] fields) {
+ this.recordType = recordType;
+ this.fields = new ArrayList<IJObject>();
+ for (IJObject jObject : fields) {
+ this.fields.add(jObject);
+ }
+ initFieldInfo();
+ }
+
+ public JRecord(String[] fieldNames, IJObject[] fields) throws AsterixException {
+ this.recordType = getARecordType(fieldNames, fields);
+ this.fields = new ArrayList<IJObject>();
+ for (IJObject jObject : fields) {
+ this.fields.add(jObject);
+ }
+ initFieldInfo();
+ }
+
+ private ARecordType getARecordType(String[] fieldNames, IJObject[] fields) throws AsterixException {
+ IAType[] fieldTypes = new IAType[fields.length];
+ int index = 0;
+ for (IJObject jObj : fields) {
+ fieldTypes[index++] = jObj.getIAObject().getType();
+ }
+ ARecordType recordType = new ARecordType(null, fieldNames, fieldTypes, false);
+ return recordType;
+ }
+
+ private void initFieldInfo() {
+ this.openField = new ArrayList<Boolean>();
+ fieldNames = new ArrayList<String>();
+ for (String name : recordType.getFieldNames()) {
+ fieldNames.add(name);
+ openField.add(false);
+ }
+ fieldTypes = new ArrayList<IAType>();
+ for (IAType type : recordType.getFieldTypes()) {
+ fieldTypes.add(type);
+ }
+
+ }
+
+ private IAObject[] getIAObjectArray(List<IJObject> fields) {
+ IAObject[] retValue = new IAObject[fields.size()];
+ int index = 0;
+ for (IJObject jObject : fields) {
+ retValue[index++] = getIAObject(jObject);
+ }
+ return retValue;
+ }
+
+ private IAObject getIAObject(IJObject jObject) {
+ IAObject retVal = null;
+ switch (jObject.getTypeTag()) {
+ case RECORD:
+ ARecordType recType = ((JRecord) jObject).getRecordType();
+ IAObject[] fields = new IAObject[((JRecord) jObject).getFields().size()];
+ int index = 0;
+ for (IJObject field : ((JRecord) jObject).getFields()) {
+ fields[index++] = getIAObject(field);
+ }
+ retVal = new AMutableRecord(recType, fields);
+ default:
+ retVal = jObject.getIAObject();
+ break;
+ }
+ return retVal;
+ }
+
+ public void addField(String fieldName, IJObject fieldValue) {
+ int pos = getFieldPosByName(fieldName);
+ if (pos >= 0) {
+ throw new IllegalArgumentException("field already defined");
+ }
+ numFieldsAdded++;
+ fields.add(fieldValue);
+ fieldNames.add(fieldName);
+ fieldTypes.add(fieldValue.getIAObject().getType());
+ openField.add(true);
+ }
+
+ public IJObject getValueByName(String fieldName) throws AsterixException, IOException {
+ int fieldPos = getFieldPosByName(fieldName);
+ if (fieldPos < 0) {
+ throw new AsterixException("unknown field: " + fieldName);
+ }
+ return fields.get(fieldPos);
+ }
+
+ public void setValueAtPos(int pos, IJObject jtype) {
+ fields.set(pos, jtype);
+ }
+
+ public void setValue(AMutableRecord mutableRecord) {
+ this.value = mutableRecord;
+ this.recordType = mutableRecord.getType();
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return recordType.getTypeTag();
+ }
+
+ public void setField(String fieldName, IJObject fieldValue) {
+ int pos = getFieldPosByName(fieldName);
+ fields.set(pos, fieldValue);
+ }
+
+ private int getFieldPosByName(String fieldName) {
+ int index = 0;
+ for (String name : fieldNames) {
+ if (name.equals(fieldName)) {
+ return index;
+ }
+ index++;
+ }
+ return -1;
+ }
+
+ public ARecordType getRecordType() {
+ return recordType;
+ }
+
+ public List<IJObject> getFields() {
+ return fields;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ if (value == null || numFieldsAdded > 0) {
+ value = new AMutableRecord(recordType, getIAObjectArray(fields));
+ }
+ return value;
+ }
+
+ public void close() {
+ if (numFieldsAdded > 0) {
+ int totalFields = fieldNames.size();
+ for (int i = 0; i < numFieldsAdded; i++) {
+ fieldNames.remove(totalFields - 1 - i);
+ fieldTypes.remove(totalFields - 1 - i);
+ fields.remove(totalFields - 1 - i);
+ }
+ numFieldsAdded = 0;
+ }
+ }
+
+ public List<Boolean> getOpenField() {
+ return openField;
+ }
+
+ public List<String> getFieldNames() {
+ return fieldNames;
+ }
+
+ public List<IAType> getFieldTypes() {
+ return fieldTypes;
+ }
+
+ }
+
+ public static class ByteArrayAccessibleInputStream extends ByteArrayInputStream {
+
+ public ByteArrayAccessibleInputStream(byte[] buf, int offset, int length) {
+ super(buf, offset, length);
+ }
+
+ public void setContent(byte[] buf, int offset, int length) {
+ this.buf = buf;
+ this.pos = offset;
+ this.count = Math.min(offset + length, buf.length);
+ this.mark = offset;
+ }
+
+ public byte[] getArray() {
+ return buf;
+ }
+
+ public int getPosition() {
+ return pos;
+ }
+
+ public int getCount() {
+ return count;
+ }
+
+ }
+
+ public static class ByteArrayAccessibleDataInputStream extends DataInputStream {
+
+ public ByteArrayAccessibleDataInputStream(ByteArrayAccessibleInputStream in) {
+ super(in);
+ }
+
+ public ByteArrayAccessibleInputStream getInputStream() {
+ return (ByteArrayAccessibleInputStream) in;
+ }
+
+ }
+}
\ No newline at end of file
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JTypeTag.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JTypeTag.java
new file mode 100644
index 0000000..3b686f6
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JTypeTag.java
@@ -0,0 +1,12 @@
+package edu.uci.ics.asterix.external.library.java;
+
+public enum JTypeTag {
+
+ INT,
+ STRING,
+ LONG,
+ DOUBLE,
+ FLOAT,
+ LIST,
+ OBJECT
+}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JTypes.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JTypes.java
new file mode 100644
index 0000000..f5f8045
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/library/java/JTypes.java
@@ -0,0 +1,404 @@
+package edu.uci.ics.asterix.external.library.java;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.ARecordSerializerDeserializer;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AStringSerializerDeserializer;
+import edu.uci.ics.asterix.external.library.JTypeObjectFactory;
+import edu.uci.ics.asterix.om.base.AInt32;
+import edu.uci.ics.asterix.om.base.AMutableInt32;
+import edu.uci.ics.asterix.om.base.AMutableOrderedList;
+import edu.uci.ics.asterix.om.base.AMutableRecord;
+import edu.uci.ics.asterix.om.base.AMutableString;
+import edu.uci.ics.asterix.om.base.ARecord;
+import edu.uci.ics.asterix.om.base.AString;
+import edu.uci.ics.asterix.om.base.IAObject;
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.AUnionType;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.EnumDeserializer;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
+import edu.uci.ics.asterix.om.util.container.IObjectPool;
+import edu.uci.ics.asterix.om.util.container.ListObjectPool;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.NotImplementedException;
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+
+public class JTypes {
+
+ public static final class JInt implements IJObject {
+
+ private AMutableInt32 value = new AMutableInt32(0);
+
+ public void setValue(int v) {
+ this.value.setValue(v);
+ }
+
+ public int getValue() {
+ return value.getIntegerValue().intValue();
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return value.getType().getTypeTag();
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return value;
+ }
+ }
+
+ public static final class JString implements IJObject {
+
+ private AMutableString value = new AMutableString("");
+
+ public void setValue(String v) {
+ this.value.setValue(v);
+ }
+
+ public String getValue() {
+ return value.getStringValue();
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return value.getType().getTypeTag();
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return value;
+ }
+ }
+
+ public static final class JList implements IJObject {
+
+ private AOrderedListType listType;
+ private AMutableOrderedList value;
+
+ public JList(AOrderedListType listType) {
+ this.listType = listType;
+ this.value = new AMutableOrderedList(listType);
+ }
+
+ public void add(IJObject element) {
+ value.add(element.getIAObject());
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return value.getType().getTypeTag();
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ return value;
+ }
+ }
+
+ public static final class JRecord implements IJObject {
+
+ private AMutableRecord value;
+ private byte[] recordBytes;
+ private ARecordType recordType;
+ private List<IJObject> fields;
+ private List<String> fieldNames;
+ private List<IAType> fieldTypes;
+ private final IObjectPool<IJObject, IAType> objectPool = new ListObjectPool<IJObject, IAType>(
+ new JTypeObjectFactory());
+ private int numFieldsAdded = 0;
+ private List<Boolean> openField;
+
+ private final static byte SER_NULL_TYPE_TAG = ATypeTag.NULL.serialize();
+ private final static byte SER_RECORD_TYPE_TAG = ATypeTag.RECORD.serialize();
+
+ public JRecord(ARecordType recordType) {
+ this.recordType = recordType;
+ this.fields = new ArrayList<IJObject>();
+ initFieldInfo();
+ }
+
+ public JRecord(ARecordType recordType, IJObject[] fields) {
+ this.recordType = recordType;
+ this.fields = new ArrayList<IJObject>();
+ for (IJObject jObject : fields) {
+ this.fields.add(jObject);
+ }
+ initFieldInfo();
+ }
+
+ private void initFieldInfo() {
+ this.openField = new ArrayList<Boolean>();
+ fieldNames = new ArrayList<String>();
+ for (String name : recordType.getFieldNames()) {
+ fieldNames.add(name);
+ openField.add(false);
+ }
+ fieldTypes = new ArrayList<IAType>();
+ for (IAType type : recordType.getFieldTypes()) {
+ fieldTypes.add(type);
+ }
+
+ }
+
+ private IAObject[] getIAObjectArray(List<IJObject> fields) {
+ IAObject[] retValue = new IAObject[fields.size()];
+ int index = 0;
+ for (IJObject jObject : fields) {
+ retValue[index++] = getIAObject(jObject);
+ }
+ return retValue;
+ }
+
+ private IAObject getIAObject(IJObject jObject) {
+ IAObject retVal = null;
+ switch (jObject.getTypeTag()) {
+ case INT32:
+ case STRING:
+ retVal = jObject.getIAObject();
+ break;
+ case RECORD:
+ ARecordType recType = ((JRecord) jObject).getRecordType();
+ IAObject[] fields = new IAObject[((JRecord) jObject).getFields().size()];
+ int index = 0;
+ for (IJObject field : ((JRecord) jObject).getFields()) {
+ fields[index++] = getIAObject(field);
+ }
+
+ retVal = new AMutableRecord(recType, fields);
+ }
+ return retVal;
+ }
+
+ public void addField(String fieldName, IJObject fieldValue) {
+ int pos = getFieldPosByName(fieldName);
+ if (pos >= 0) {
+ throw new IllegalArgumentException("field already defined");
+ }
+ numFieldsAdded++;
+ fields.add(fieldValue);
+ fieldNames.add(fieldName);
+ fieldTypes.add(fieldValue.getIAObject().getType());
+ openField.add(true);
+ }
+
+ public IJObject getValueByName(String fieldName) throws AsterixException, IOException {
+ int fieldPos = getFieldPosByName(fieldName);
+
+ if (recordBytes == null) {
+ IJObject jtype = getJObject(value.getValueByPos(fieldPos));
+ fields.set(fieldPos, jtype);
+ return jtype;
+ }
+
+ if (recordBytes[0] == SER_NULL_TYPE_TAG) {
+ return null;
+ }
+
+ if (recordBytes[0] != SER_RECORD_TYPE_TAG) {
+ throw new AsterixException("Field accessor is not defined for values of type"
+ + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(recordBytes[0]));
+ }
+
+ int fieldValueOffset = ARecordSerializerDeserializer.getFieldOffsetById(recordBytes, fieldPos,
+ getNullBitMapSize(), recordType.isOpen());
+
+ if (fieldValueOffset < 0) {
+ return null;
+ }
+
+ IAType fieldValueType = recordType.getFieldTypes()[fieldPos];
+ ATypeTag fieldValueTypeTag = null;
+ int fieldValueLength = 0;
+ if (fieldValueType.getTypeTag().equals(ATypeTag.UNION)) {
+ if (NonTaggedFormatUtil.isOptionalField((AUnionType) fieldValueType)) {
+ fieldValueTypeTag = ((AUnionType) fieldValueType).getUnionList()
+ .get(NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST).getTypeTag();
+ fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(recordBytes, fieldValueOffset,
+ fieldValueTypeTag, false);
+ // out.writeByte(fieldValueTypeTag.serialize());
+ } else {
+ // union .. the general case
+ throw new NotImplementedException();
+ }
+ } else {
+ fieldValueTypeTag = fieldValueType.getTypeTag();
+ fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(recordBytes, fieldValueOffset,
+ fieldValueTypeTag, false);
+ // out.writeByte(fieldValueTypeTag.serialize());
+ }
+
+ IJObject fieldValue = getJType(fieldValueTypeTag, recordBytes, fieldValueOffset, fieldValueLength, fieldPos);
+ fields.set(fieldPos, fieldValue);
+ return fieldValue;
+ }
+
+ public void setValue(byte[] recordBytes) {
+ this.recordBytes = recordBytes;
+ }
+
+ public void setValueAtPos(int pos, IJObject jtype) {
+ fields.set(pos, jtype);
+ }
+
+ public void setValue(AMutableRecord mutableRecord) {
+ this.value = mutableRecord;
+ this.recordType = mutableRecord.getType();
+ }
+
+ @Override
+ public ATypeTag getTypeTag() {
+ return recordType.getTypeTag();
+ }
+
+ public void setField(String fieldName, IJObject fieldValue) {
+ int pos = getFieldPosByName(fieldName);
+ switch (fields.get(pos).getTypeTag()) {
+ case INT32:
+ ((JInt) fields.get(pos)).setValue(((AMutableInt32) fieldValue.getIAObject()).getIntegerValue()
+ .intValue());
+ break;
+ case STRING:
+ ((JString) fields.get(pos)).setValue(((AMutableString) fieldValue.getIAObject()).getStringValue());
+ break;
+ case RECORD:
+ ((JRecord) fields.get(pos)).setValue(((AMutableRecord) fieldValue));
+ break;
+ }
+ }
+
+ private int getFieldPosByName(String fieldName) {
+ int index = 0;
+ for (String name : fieldNames) {
+ if (name.equals(fieldName)) {
+ return index;
+ }
+ index++;
+ }
+ return -1;
+ }
+
+ private IJObject getJType(ATypeTag typeTag, byte[] argument, int offset, int len, int fieldIndex)
+ throws HyracksDataException {
+ IJObject jObject;
+ switch (typeTag) {
+ case INT32: {
+ int v = valueFromBytes(argument, offset, len);
+ jObject = objectPool.allocate(BuiltinType.AINT32);
+ ((JInt) jObject).setValue(v);
+ break;
+
+ }
+ case STRING: {
+ String v = AStringSerializerDeserializer.INSTANCE.deserialize(
+ new DataInputStream(new ByteArrayInputStream(argument, offset, len))).getStringValue();
+ jObject = objectPool.allocate(BuiltinType.ASTRING);
+ ((JString) jObject).setValue(v);
+ break;
+ }
+ case RECORD:
+ ARecordType fieldRecordType = (ARecordType) recordType.getFieldTypes()[fieldIndex];
+ jObject = objectPool.allocate(fieldRecordType);
+ byte[] recBytes = new byte[len];
+ System.arraycopy(argument, offset, recBytes, 0, len);
+ ((JRecord) jObject).setValue(argument);
+ break;
+ default:
+ throw new IllegalStateException("Argument type: " + typeTag);
+ }
+ return jObject;
+ }
+
+ private IJObject getJObject(IAObject iaobject) throws HyracksDataException {
+ ATypeTag typeTag = iaobject.getType().getTypeTag();
+ IJObject jtype;
+ switch (typeTag) {
+ case INT32: {
+ int v = ((AInt32) iaobject).getIntegerValue().intValue();
+ jtype = new JInt();
+ ((JInt) jtype).setValue(v);
+ break;
+ }
+ case STRING: {
+ jtype = new JString();
+ ((JString) jtype).setValue(((AString) iaobject).getStringValue());
+ break;
+ }
+ case RECORD:
+ ARecordType fieldRecordType = ((ARecord) iaobject).getType();
+ jtype = new JRecord(fieldRecordType);
+ ((JRecord) jtype).setValue((AMutableRecord) iaobject);
+ break;
+ default:
+ throw new IllegalStateException("Argument type: " + typeTag);
+ }
+ return jtype;
+ }
+
+ private static int valueFromBytes(byte[] bytes, int offset, int length) {
+ return ((bytes[offset] & 0xff) << 24) + ((bytes[offset + 1] & 0xff) << 16)
+ + ((bytes[offset + 2] & 0xff) << 8) + ((bytes[offset + 3] & 0xff) << 0);
+ }
+
+ public ARecordType getRecordType() {
+ return recordType;
+ }
+
+ public List<IJObject> getFields() {
+ return fields;
+ }
+
+ @Override
+ public IAObject getIAObject() {
+ if (value == null || numFieldsAdded > 0) {
+ value = new AMutableRecord(recordType, getIAObjectArray(fields));
+ }
+ return value;
+ }
+
+ public void close() {
+ objectPool.reset();
+ if (numFieldsAdded > 0) {
+ int totalFields = fieldNames.size();
+ for (int i = 0; i < numFieldsAdded; i++) {
+ fieldNames.remove(totalFields - 1 - i);
+ fieldTypes.remove(totalFields - 1 - i);
+ fields.remove(totalFields - 1 - i);
+ }
+ numFieldsAdded = 0;
+ }
+ }
+
+ private int getNullBitMapSize() {
+ int nullBitmapSize = 0;
+ if (NonTaggedFormatUtil.hasNullableField(recordType)) {
+ nullBitmapSize = (int) Math.ceil(recordType.getFieldNames().length / 8.0);
+ } else {
+ nullBitmapSize = 0;
+ }
+ return nullBitmapSize;
+ }
+
+ public List<Boolean> getOpenField() {
+ return openField;
+ }
+
+ public List<String> getFieldNames() {
+ return fieldNames;
+ }
+
+ public List<IAType> getFieldTypes() {
+ return fieldTypes;
+ }
+
+ }
+}
\ No newline at end of file
diff --git a/asterix-external-data/src/main/resources/schema/library.xsd b/asterix-external-data/src/main/resources/schema/library.xsd
new file mode 100644
index 0000000..3dc4659
--- /dev/null
+++ b/asterix-external-data/src/main/resources/schema/library.xsd
@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="ISO-8859-1" ?>
+<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema" xmlns:lib="library" targetNamespace="library" elementFormDefault="qualified">
+
+<!-- definition of simple types -->
+<xs:element name="language" type="xs:string"/>
+<xs:element name="name" type="xs:string"/>
+<xs:element name="arguments" type="xs:string"/>
+<xs:element name="return_type" type="xs:string"/>
+<xs:element name="function_type" type="xs:string"/>
+<xs:element name="definition" type="xs:string"/>
+
+<!-- definition of complex elements -->
+<xs:element name="function">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element ref="lib:name"/>
+ <xs:element ref="lib:function_type"/>
+ <xs:element ref="lib:arguments"/>
+ <xs:element ref="lib:return_type"/>
+ <xs:element ref="lib:definition"/>
+ </xs:sequence>
+ </xs:complexType>
+</xs:element>
+
+<xs:element name="functions">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element ref="lib:function" maxOccurs="unbounded"/>
+ </xs:sequence>
+ </xs:complexType>
+</xs:element>
+
+<xs:element name="library">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element ref="lib:language"/>
+ <xs:element ref="lib:functions" minOccurs="0"/>
+ </xs:sequence>
+ </xs:complexType>
+</xs:element>
+
+</xs:schema>
diff --git a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/CommandHandler.java b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/CommandHandler.java
index 7fc3884..21cec69 100644
--- a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/CommandHandler.java
+++ b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/CommandHandler.java
@@ -52,6 +52,12 @@
case CONFIGURE:
cmd = new ConfigureCommand();
break;
+ case INSTALL:
+ cmd = new InstallCommand();
+ break;
+ case UNINSTALL:
+ cmd = new UninstallCommand();
+ break;
case SHUTDOWN:
cmd = new ShutdownCommand();
break;
diff --git a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/HelpCommand.java b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/HelpCommand.java
index a9864a3..f0e2fcc 100644
--- a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/HelpCommand.java
+++ b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/HelpCommand.java
@@ -55,6 +55,12 @@
case VALIDATE:
helpMessage = new ValidateCommand().getUsageDescription();
break;
+ case INSTALL:
+ helpMessage = new InstallCommand().getUsageDescription();
+ break;
+ case UNINSTALL:
+ helpMessage = new UninstallCommand().getUsageDescription();
+ break;
default:
helpMessage = "Unknown command " + command;
}
diff --git a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/ICommand.java b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/ICommand.java
index 979c414..be64aa4 100644
--- a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/ICommand.java
+++ b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/ICommand.java
@@ -27,6 +27,8 @@
ALTER,
VALIDATE,
CONFIGURE,
+ INSTALL,
+ UNINSTALL,
SHUTDOWN,
HELP
}
diff --git a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/InstallCommand.java b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/InstallCommand.java
new file mode 100644
index 0000000..936efcb
--- /dev/null
+++ b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/InstallCommand.java
@@ -0,0 +1,71 @@
+/*
+ * Copyright 2009-2012 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.installer.command;
+
+import org.kohsuke.args4j.Option;
+
+import edu.uci.ics.asterix.event.schema.pattern.Patterns;
+import edu.uci.ics.asterix.installer.driver.InstallerDriver;
+import edu.uci.ics.asterix.installer.driver.InstallerUtil;
+import edu.uci.ics.asterix.installer.events.PatternCreator;
+import edu.uci.ics.asterix.installer.model.AsterixInstance;
+import edu.uci.ics.asterix.installer.model.AsterixInstance.State;
+
+public class InstallCommand extends AbstractCommand {
+
+ @Override
+ protected void execCommand() throws Exception {
+ InstallerDriver.initConfig();
+ InstallConfig installConfig = ((InstallConfig) config);
+ String instanceName = installConfig.name;
+ AsterixInstance instance = InstallerUtil.validateAsterixInstanceExists(instanceName, State.INACTIVE);
+ PatternCreator pc = new PatternCreator();
+ Patterns patterns = pc.getLibraryInstallPattern(instance, installConfig.dataverseName,
+ installConfig.libraryName, installConfig.libraryPath);
+ InstallerUtil.getEventrixClient(instance.getCluster()).submit(patterns);
+ LOGGER.info("Installed library " + installConfig.libraryName);
+ }
+
+ @Override
+ protected CommandConfig getCommandConfig() {
+ return new InstallConfig();
+ }
+
+ @Override
+ protected String getUsageDescription() {
+ return "Installs a library to an asterix instance." + "\n" + "Arguments/Options\n"
+ + "-n Name of Asterix Instance\n"
+ + "-d Name of the dataverse under which the library will be installed\n" + "-l Name of the library\n"
+ + "-p Path to library zip bundle";
+
+ }
+
+}
+
+class InstallConfig extends CommandConfig {
+
+ @Option(name = "-n", required = true, usage = "Name of Asterix Instance")
+ public String name;
+
+ @Option(name = "-d", required = true, usage = "Name of the dataverse under which the library will be installed")
+ public String dataverseName;
+
+ @Option(name = "-l", required = true, usage = "Name of the library")
+ public String libraryName;
+
+ @Option(name = "-p", required = true, usage = "Path to library zip bundle")
+ public String libraryPath;
+
+}
diff --git a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/UninstallCommand.java b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/UninstallCommand.java
new file mode 100644
index 0000000..70b4d8c
--- /dev/null
+++ b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/command/UninstallCommand.java
@@ -0,0 +1,71 @@
+/*
+ * Copyright 2009-2012 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.installer.command;
+
+import org.kohsuke.args4j.Option;
+
+import edu.uci.ics.asterix.event.schema.pattern.Patterns;
+import edu.uci.ics.asterix.installer.driver.InstallerDriver;
+import edu.uci.ics.asterix.installer.driver.InstallerUtil;
+import edu.uci.ics.asterix.installer.events.PatternCreator;
+import edu.uci.ics.asterix.installer.model.AsterixInstance;
+import edu.uci.ics.asterix.installer.model.AsterixInstance.State;
+import edu.uci.ics.asterix.installer.service.ILookupService;
+import edu.uci.ics.asterix.installer.service.ServiceProvider;
+
+public class UninstallCommand extends AbstractCommand {
+
+ @Override
+ protected void execCommand() throws Exception {
+ InstallerDriver.initConfig();
+ UninstallConfig uninstallConfig = ((UninstallConfig) config);
+ String instanceName = uninstallConfig.name;
+ InstallerUtil.validateAsterixInstanceExists(instanceName, State.INACTIVE);
+ ILookupService lookupService = ServiceProvider.INSTANCE.getLookupService();
+ AsterixInstance instance = lookupService.getAsterixInstance(instanceName);
+ PatternCreator pc = new PatternCreator();
+ Patterns patterns = pc.getLibraryUninstallPattern(instance, uninstallConfig.dataverseName,
+ uninstallConfig.libraryName);
+ InstallerUtil.getEventrixClient(instance.getCluster()).submit(patterns);
+ LOGGER.info("Uninstalled library " + uninstallConfig.libraryName);
+ }
+
+ @Override
+ protected CommandConfig getCommandConfig() {
+ return new UninstallConfig();
+ }
+
+ @Override
+ protected String getUsageDescription() {
+ return "Uninstalls a library from an asterix instance." + "\n" + "Arguments/Options\n"
+ + "-n Name of Asterix Instance\n"
+ + "-d Name of the dataverse under which the library will be installed\n" + "-l Name of the library\n"
+ + "-l Name of the library";
+ }
+
+}
+
+class UninstallConfig extends CommandConfig {
+
+ @Option(name = "-n", required = true, usage = "Name of Asterix Instance")
+ public String name;
+
+ @Option(name = "-d", required = true, usage = "Name of the dataverse under which the library will be installed")
+ public String dataverseName;
+
+ @Option(name = "-l", required = true, usage = "Name of the library")
+ public String libraryName;
+
+}
diff --git a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/driver/InstallerDriver.java b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/driver/InstallerDriver.java
index 8083427..ff3bf89 100644
--- a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/driver/InstallerDriver.java
+++ b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/driver/InstallerDriver.java
@@ -30,133 +30,114 @@
public class InstallerDriver {
- public static final String MANAGIX_INTERNAL_DIR = ".installer";
- public static final String MANAGIX_EVENT_DIR = MANAGIX_INTERNAL_DIR
- + File.separator + "eventrix";
- public static final String MANAGIX_EVENT_SCRIPTS_DIR = MANAGIX_INTERNAL_DIR
- + File.separator + "eventrix" + File.separator + "scripts";
- public static final String ASTERIX_DIR = "asterix";
- public static final String EVENTS_DIR = "events";
+ public static final String MANAGIX_INTERNAL_DIR = ".installer";
+ public static final String MANAGIX_EVENT_DIR = MANAGIX_INTERNAL_DIR + File.separator + "eventrix";
+ public static final String MANAGIX_EVENT_SCRIPTS_DIR = MANAGIX_INTERNAL_DIR + File.separator + "eventrix"
+ + File.separator + "scripts";
+ public static final String ASTERIX_DIR = "asterix";
+ public static final String EVENTS_DIR = "events";
- private static final Logger LOGGER = Logger.getLogger(InstallerDriver.class
- .getName());
- public static final String ENV_MANAGIX_HOME = "MANAGIX_HOME";
- public static final String MANAGIX_CONF_XML = "conf" + File.separator
- + "managix-conf.xml";
+ private static final Logger LOGGER = Logger.getLogger(InstallerDriver.class.getName());
+ public static final String ENV_MANAGIX_HOME = "MANAGIX_HOME";
+ public static final String MANAGIX_CONF_XML = "conf" + File.separator + "managix-conf.xml";
- private static Configuration conf;
- private static String managixHome;
- private static String asterixZip;
+ private static Configuration conf;
+ private static String managixHome;
+ private static String asterixZip;
- public static String getAsterixZip() {
- return asterixZip;
- }
+ public static String getAsterixZip() {
+ return asterixZip;
+ }
- public static Configuration getConfiguration() {
- return conf;
- }
+ public static Configuration getConfiguration() {
+ return conf;
+ }
- public static void initConfig() throws Exception {
- File configFile = new File(managixHome + File.separator
- + MANAGIX_CONF_XML);
- JAXBContext configCtx = JAXBContext.newInstance(Configuration.class);
- Unmarshaller unmarshaller = configCtx.createUnmarshaller();
- conf = (Configuration) unmarshaller.unmarshal(configFile);
- asterixZip = initBinary("asterix-server");
+ public static void initConfig() throws Exception {
+ File configFile = new File(managixHome + File.separator + MANAGIX_CONF_XML);
+ JAXBContext configCtx = JAXBContext.newInstance(Configuration.class);
+ Unmarshaller unmarshaller = configCtx.createUnmarshaller();
+ conf = (Configuration) unmarshaller.unmarshal(configFile);
+ asterixZip = initBinary("asterix-server");
- ILookupService lookupService = ServiceProvider.INSTANCE
- .getLookupService();
- if (!lookupService.isRunning(conf)) {
- lookupService.startService(conf);
- }
- }
+ ILookupService lookupService = ServiceProvider.INSTANCE.getLookupService();
+ if (!lookupService.isRunning(conf)) {
+ lookupService.startService(conf);
+ }
+ }
- private static String initBinary(final String fileNamePattern) {
- String asterixDir = InstallerDriver.getAsterixDir();
- File file = new File(asterixDir);
- File[] zipFiles = file.listFiles(new FileFilter() {
- public boolean accept(File arg0) {
- return arg0.getAbsolutePath().contains(fileNamePattern)
- && arg0.isFile();
- }
- });
- if (zipFiles.length == 0) {
- String msg = " Binary not found at " + asterixDir;
- LOGGER.log(Level.FATAL, msg);
- throw new IllegalStateException(msg);
- }
- if (zipFiles.length > 1) {
- String msg = " Multiple binaries found at " + asterixDir;
- LOGGER.log(Level.FATAL, msg);
- throw new IllegalStateException(msg);
- }
+ private static String initBinary(final String fileNamePattern) {
+ String asterixDir = InstallerDriver.getAsterixDir();
+ File file = new File(asterixDir);
+ File[] zipFiles = file.listFiles(new FileFilter() {
+ public boolean accept(File arg0) {
+ return arg0.getAbsolutePath().contains(fileNamePattern) && arg0.isFile();
+ }
+ });
+ if (zipFiles.length == 0) {
+ String msg = " Binary not found at " + asterixDir;
+ LOGGER.log(Level.FATAL, msg);
+ throw new IllegalStateException(msg);
+ }
+ if (zipFiles.length > 1) {
+ String msg = " Multiple binaries found at " + asterixDir;
+ LOGGER.log(Level.FATAL, msg);
+ throw new IllegalStateException(msg);
+ }
- return zipFiles[0].getAbsolutePath();
- }
+ return zipFiles[0].getAbsolutePath();
+ }
- public static String getManagixHome() {
- return managixHome;
- }
+ public static String getManagixHome() {
+ return managixHome;
+ }
- public static String getAsterixDir() {
- return managixHome + File.separator + ASTERIX_DIR;
- }
+ public static String getAsterixDir() {
+ return managixHome + File.separator + ASTERIX_DIR;
+ }
- public static void main(String args[]) {
- try {
- if (args.length != 0) {
- managixHome = System.getenv(ENV_MANAGIX_HOME);
- CommandHandler cmdHandler = new CommandHandler();
- cmdHandler.processCommand(args);
- } else {
- printUsage();
- }
- } catch (IllegalArgumentException iae) {
- LOGGER.error("Unknown command");
- printUsage();
- } catch (Exception e) {
- LOGGER.error(e.getMessage());
- if (e.getMessage() == null || e.getMessage().length() == 0) {
- e.printStackTrace();
- }
- }
- }
+ public static void main(String args[]) {
+ try {
+ if (args.length != 0) {
+ managixHome = System.getenv(ENV_MANAGIX_HOME);
+ CommandHandler cmdHandler = new CommandHandler();
+ cmdHandler.processCommand(args);
+ } else {
+ printUsage();
+ }
+ } catch (IllegalArgumentException iae) {
+ LOGGER.error("Unknown command");
+ printUsage();
+ } catch (Exception e) {
+ LOGGER.error(e.getMessage());
+ if (e.getMessage() == null || e.getMessage().length() == 0) {
+ e.printStackTrace();
+ }
+ }
+ }
- private static void printUsage() {
- StringBuffer buffer = new StringBuffer("managix <command> <options>"
- + "\n");
- buffer.append("Commands" + "\n");
- buffer.append("create " + ":" + " Creates a new asterix instance"
- + "\n");
- buffer.append("delete " + ":" + " Deletes an asterix instance" + "\n");
- buffer.append("start " + ":" + " Starts an asterix instance" + "\n");
- buffer.append("stop " + ":"
- + " Stops an asterix instance that is in ACTIVE state" + "\n");
- buffer.append("backup " + ":"
- + " Creates a back up for an existing asterix instance" + "\n");
- buffer.append("restore " + ":" + " Restores an asterix instance"
- + "\n");
- buffer.append("describe " + ":"
- + " Describes an existing asterix instance" + "\n");
- buffer.append("validate " + ":"
- + " Validates the installer/cluster configuration" + "\n");
- buffer.append("configure"
- + ":"
- + " Auto-generate configuration for local psedu-distributed Asterix instance"
- + "\n");
- buffer.append("shutdown " + ":" + " Shutdown the installer service"
- + "\n");
- buffer.append("validate " + ":"
- + " Validates the installer/cluster configuration" + "\n");
- buffer.append("configure"
- + ":"
- + " Auto-generate configuration for local psedu-distributed Asterix instance"
- + "\n");
- buffer.append("shutdown " + ":" + " Shutdown the installer service"
- + "\n");
- buffer.append("help " + ":"
- + " Provides usage description of a command" + "\n");
+ private static void printUsage() {
+ StringBuffer buffer = new StringBuffer("managix <command> <options>" + "\n");
+ buffer.append("Commands" + "\n");
+ buffer.append("create " + ":" + " Creates a new asterix instance" + "\n");
+ buffer.append("delete " + ":" + " Deletes an asterix instance" + "\n");
+ buffer.append("start " + ":" + " Starts an asterix instance" + "\n");
+ buffer.append("stop " + ":" + " Stops an asterix instance that is in ACTIVE state" + "\n");
+ buffer.append("backup " + ":" + " Creates a back up for an existing asterix instance" + "\n");
+ buffer.append("restore " + ":" + " Restores an asterix instance" + "\n");
+ buffer.append("describe " + ":" + " Describes an existing asterix instance" + "\n");
+ buffer.append("validate " + ":" + " Validates the installer/cluster configuration" + "\n");
+ buffer.append("configure" + ":" + " Auto-generate configuration for local psedu-distributed Asterix instance"
+ + "\n");
+ buffer.append("install " + ":" + " Installs a library to an asterix instance" + "\n");
+ buffer.append("uninstall" + ":" + " Uninstalls a library from an asterix instance" + "\n");
+ buffer.append("shutdown " + ":" + " Shutdown the installer service" + "\n");
+ buffer.append("validate " + ":" + " Validates the installer/cluster configuration" + "\n");
+ buffer.append("configure" + ":" + " Auto-generate configuration for local psedu-distributed Asterix instance"
+ + "\n");
+ buffer.append("shutdown " + ":" + " Shutdown the installer service" + "\n");
+ buffer.append("help " + ":" + " Provides usage description of a command" + "\n");
- LOGGER.info(buffer.toString());
- }
+ LOGGER.info(buffer.toString());
+ }
}
diff --git a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/events/PatternCreator.java b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/events/PatternCreator.java
index f6aeafe..ec3b09f 100644
--- a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/events/PatternCreator.java
+++ b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/events/PatternCreator.java
@@ -17,6 +17,7 @@
import java.io.File;
import java.util.ArrayList;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Set;
@@ -356,6 +357,85 @@
return patterns;
}
+ public Patterns getLibraryInstallPattern(AsterixInstance instance, String dataverse, String libraryName,
+ String libraryPath) throws Exception {
+ List<Pattern> patternList = new ArrayList<Pattern>();
+ Cluster cluster = instance.getCluster();
+ Nodeid nodeid = new Nodeid(new Value(null, EventDriver.CLIENT_NODE.getId()));
+ String username = cluster.getUsername() != null ? cluster.getUsername() : System.getProperty("user.name");
+ String workingDir = cluster.getWorkingDir().getDir();
+ String destDir = workingDir + File.separator + "library" + File.separator + dataverse + File.separator
+ + libraryName;
+ String fileToTransfer = new File(libraryPath).getAbsolutePath();
+
+ Iterator<Node> installTargets = cluster.getNode().iterator();
+ Node installNode = installTargets.next();
+ String destinationIp = installNode.getClusterIp();
+ String pargs = username + " " + fileToTransfer + " " + destinationIp + " " + destDir + " " + "unpack";
+ Event event = new Event("file_transfer", nodeid, pargs);
+ Pattern p = new Pattern(null, 1, null, event);
+ patternList.add(p);
+
+ if (!cluster.getWorkingDir().isNFS()) {
+ while (installTargets.hasNext()) {
+ Node node = installTargets.next();
+ pargs = username + " " + fileToTransfer + " " + node.getClusterIp() + " " + destDir + " " + "unpack";
+ event = new Event("file_transfer", nodeid, pargs);
+ p = new Pattern(null, 1, null, event);
+ patternList.add(p);
+ }
+
+ pargs = username + " " + fileToTransfer + " " + cluster.getMasterNode().getClusterIp() + " " + destDir
+ + " " + "unpack";
+ event = new Event("file_transfer", nodeid, pargs);
+ p = new Pattern(null, 1, null, event);
+ patternList.add(p);
+ }
+ return new Patterns(patternList);
+ }
+
+ public Patterns getLibraryUninstallPattern(AsterixInstance instance, String dataverse, String libraryName)
+ throws Exception {
+ List<Pattern> patternList = new ArrayList<Pattern>();
+ Cluster cluster = instance.getCluster();
+ String workingDir = cluster.getWorkingDir().getDir();
+ String destFile = dataverse + "." + libraryName;
+ String pargs = workingDir + File.separator + "uninstall" + " " + destFile;
+
+ String metadataNodeId = instance.getMetadataNodeId();
+ Nodeid nodeid = new Nodeid(new Value(null, metadataNodeId));
+ Event event = new Event("file_create", nodeid, pargs);
+ Pattern p = new Pattern(null, 1, null, event);
+ patternList.add(p);
+
+ Iterator<Node> uninstallTargets = cluster.getNode().iterator();
+ String libDir = workingDir + File.separator + "library" + File.separator + dataverse + File.separator
+ + libraryName;
+ Node uninstallNode = uninstallTargets.next();
+ nodeid = new Nodeid(new Value(null, uninstallNode.getId()));
+ event = new Event("file_delete", nodeid, libDir);
+ p = new Pattern(null, 1, null, event);
+ patternList.add(p);
+ pargs = libDir;
+
+ if (!cluster.getWorkingDir().isNFS()) {
+ while (uninstallTargets.hasNext()) {
+ uninstallNode = uninstallTargets.next();
+ nodeid = new Nodeid(new Value(null, uninstallNode.getId()));
+ event = new Event("file_delete", nodeid, pargs);
+ p = new Pattern(null, 1, null, event);
+ patternList.add(p);
+ }
+
+ nodeid = new Nodeid(new Value(null, cluster.getMasterNode().getId()));
+ event = new Event("file_delete", nodeid, pargs);
+ p = new Pattern(null, 1, null, event);
+ patternList.add(p);
+
+ }
+ return new Patterns(patternList);
+ }
+
private Patterns createRemoveAsterixRootMetadata(AsterixInstance instance) throws Exception {
List<Pattern> patternList = new ArrayList<Pattern>();
Cluster cluster = instance.getCluster();
diff --git a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/service/ZooKeeperService.java b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/service/ZooKeeperService.java
index b87d721..81adee0 100644
--- a/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/service/ZooKeeperService.java
+++ b/asterix-installer/src/main/java/edu/uci/ics/asterix/installer/service/ZooKeeperService.java
@@ -168,7 +168,7 @@
public void removeAsterixInstance(String name) throws Exception {
if (!exists(name)) {
- throw new InstallerException("Asterix instance by name " + name + " does not exists.");
+ throw new InstallerException("Asterix instance by name " + name + " does not exists.");
}
zk.delete(ASTERIX_INSTANCE_BASE_PATH + File.separator + name, DEFAULT_NODE_VERSION);
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataCache.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataCache.java
index 0382315..4a86bca 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataCache.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataCache.java
@@ -21,13 +21,15 @@
import java.util.Map;
import edu.uci.ics.asterix.common.functions.FunctionSignature;
+import edu.uci.ics.asterix.external.dataset.adapter.AdapterIdentifier;
import edu.uci.ics.asterix.metadata.api.IMetadataEntity;
-import edu.uci.ics.asterix.metadata.entities.DatasourceAdapter;
import edu.uci.ics.asterix.metadata.entities.Dataset;
+import edu.uci.ics.asterix.metadata.entities.DatasourceAdapter;
import edu.uci.ics.asterix.metadata.entities.Datatype;
import edu.uci.ics.asterix.metadata.entities.Dataverse;
import edu.uci.ics.asterix.metadata.entities.Function;
import edu.uci.ics.asterix.metadata.entities.Index;
+import edu.uci.ics.asterix.metadata.entities.Library;
import edu.uci.ics.asterix.metadata.entities.NodeGroup;
/**
@@ -37,368 +39,407 @@
* simply ignored, i.e., updates are not not applied to the cache.
*/
public class MetadataCache {
- // Key is dataverse name.
- protected final Map<String, Dataverse> dataverses = new HashMap<String, Dataverse>();
- // Key is dataverse name. Key of value map is dataset name.
- protected final Map<String, Map<String, Dataset>> datasets = new HashMap<String, Map<String, Dataset>>();
- // Key is dataverse name. Key of value map is dataset name. Key of value map of value map is index name.
- protected final Map<String, Map<String, Map<String, Index>>> indexes = new HashMap<String, Map<String, Map<String, Index>>>();
- // Key is dataverse name. Key of value map is datatype name.
- protected final Map<String, Map<String, Datatype>> datatypes = new HashMap<String, Map<String, Datatype>>();
- // Key is dataverse name.
- protected final Map<String, NodeGroup> nodeGroups = new HashMap<String, NodeGroup>();
- // Key is function Identifier . Key of value map is function name.
- protected final Map<FunctionSignature, Function> functions = new HashMap<FunctionSignature, Function>();
- // Key is adapter dataverse. Key of value map is the adapter name
- protected final Map<String, Map<String, DatasourceAdapter>> adapters = new HashMap<String, Map<String, DatasourceAdapter>>();
+ // Key is dataverse name.
+ protected final Map<String, Dataverse> dataverses = new HashMap<String, Dataverse>();
+ // Key is dataverse name. Key of value map is dataset name.
+ protected final Map<String, Map<String, Dataset>> datasets = new HashMap<String, Map<String, Dataset>>();
+ // Key is dataverse name. Key of value map is dataset name. Key of value map
+ // of value map is index name.
+ protected final Map<String, Map<String, Map<String, Index>>> indexes = new HashMap<String, Map<String, Map<String, Index>>>();
+ // Key is dataverse name. Key of value map is datatype name.
+ protected final Map<String, Map<String, Datatype>> datatypes = new HashMap<String, Map<String, Datatype>>();
+ // Key is dataverse name.
+ protected final Map<String, NodeGroup> nodeGroups = new HashMap<String, NodeGroup>();
+ // Key is function Identifier . Key of value map is function name.
+ protected final Map<FunctionSignature, Function> functions = new HashMap<FunctionSignature, Function>();
+ // Key is adapter dataverse. Key of value map is the adapter name
+ protected final Map<AdapterIdentifier, DatasourceAdapter> adapters = new HashMap<AdapterIdentifier, DatasourceAdapter>();
+ // Key is library dataverse. Key of value map is the library name
+ protected final Map<String, Map<String, Library>> libraries = new HashMap<String, Map<String, Library>>();
- // Atomically executes all metadata operations in ctx's log.
- public void commit(MetadataTransactionContext ctx) {
- // Forward roll the operations written in ctx's log.
- int logIx = 0;
- ArrayList<MetadataLogicalOperation> opLog = ctx.getOpLog();
- try {
- for (logIx = 0; logIx < opLog.size(); logIx++) {
- doOperation(opLog.get(logIx));
- }
- } catch (Exception e) {
- // Undo operations.
- try {
- for (int i = logIx - 1; i >= 0; i--) {
- undoOperation(opLog.get(i));
- }
- } catch (Exception e2) {
- // We encountered an error in undo. This case should never
- // happen. Our only remedy to ensure cache consistency
- // is to clear everything.
- clear();
- }
- } finally {
- ctx.clear();
- }
- }
+ // Atomically executes all metadata operations in ctx's log.
+ public void commit(MetadataTransactionContext ctx) {
+ // Forward roll the operations written in ctx's log.
+ int logIx = 0;
+ ArrayList<MetadataLogicalOperation> opLog = ctx.getOpLog();
+ try {
+ for (logIx = 0; logIx < opLog.size(); logIx++) {
+ doOperation(opLog.get(logIx));
+ }
+ } catch (Exception e) {
+ // Undo operations.
+ try {
+ for (int i = logIx - 1; i >= 0; i--) {
+ undoOperation(opLog.get(i));
+ }
+ } catch (Exception e2) {
+ // We encountered an error in undo. This case should never
+ // happen. Our only remedy to ensure cache consistency
+ // is to clear everything.
+ clear();
+ }
+ } finally {
+ ctx.clear();
+ }
+ }
- public void clear() {
- synchronized (dataverses) {
- synchronized (nodeGroups) {
- synchronized (datasets) {
- synchronized (indexes) {
- synchronized (datatypes) {
- synchronized (functions) {
- synchronized (adapters) {
- dataverses.clear();
- nodeGroups.clear();
- datasets.clear();
- indexes.clear();
- datatypes.clear();
- functions.clear();
- adapters.clear();
- }
- }
- }
- }
- }
- }
- }
- }
+ public void clear() {
+ synchronized (dataverses) {
+ synchronized (nodeGroups) {
+ synchronized (datasets) {
+ synchronized (datatypes) {
+ synchronized (functions) {
+ synchronized (adapters) {
+ synchronized (libraries) {
+ dataverses.clear();
+ nodeGroups.clear();
+ datasets.clear();
+ datatypes.clear();
+ functions.clear();
+ adapters.clear();
+ libraries.clear();
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
- public Object addDataverseIfNotExists(Dataverse dataverse) {
- synchronized (dataverses) {
- synchronized (datasets) {
- synchronized (datatypes) {
- if (!dataverses.containsKey(dataverse)) {
- datasets.put(dataverse.getDataverseName(), new HashMap<String, Dataset>());
- datatypes.put(dataverse.getDataverseName(), new HashMap<String, Datatype>());
- adapters.put(dataverse.getDataverseName(), new HashMap<String, DatasourceAdapter>());
- return dataverses.put(dataverse.getDataverseName(), dataverse);
- }
- return null;
- }
- }
- }
- }
+ public Object addDataverseIfNotExists(Dataverse dataverse) {
+ synchronized (dataverses) {
+ synchronized (datasets) {
+ synchronized (datatypes) {
+ if (!dataverses.containsKey(dataverse)) {
+ datasets.put(dataverse.getDataverseName(),
+ new HashMap<String, Dataset>());
+ datatypes.put(dataverse.getDataverseName(),
+ new HashMap<String, Datatype>());
+ return dataverses.put(dataverse.getDataverseName(),
+ dataverse);
+ }
+ return null;
+ }
+ }
+ }
+ }
- public Object addDatasetIfNotExists(Dataset dataset) {
- synchronized (datasets) {
- Map<String, Dataset> m = datasets.get(dataset.getDataverseName());
- if (m == null) {
- m = new HashMap<String, Dataset>();
- datasets.put(dataset.getDataverseName(), m);
- }
- if (!m.containsKey(dataset.getDatasetName())) {
- return m.put(dataset.getDatasetName(), dataset);
- }
- return null;
- }
- }
+ public Object addDatasetIfNotExists(Dataset dataset) {
+ synchronized (datasets) {
+ Map<String, Dataset> m = datasets.get(dataset.getDataverseName());
+ if (m == null) {
+ m = new HashMap<String, Dataset>();
+ datasets.put(dataset.getDataverseName(), m);
+ }
+ if (!m.containsKey(dataset.getDatasetName())) {
+ return m.put(dataset.getDatasetName(), dataset);
+ }
+ return null;
+ }
+ }
- public Object addIndexIfNotExists(Index index) {
- synchronized (indexes) {
- Map<String, Map<String, Index>> datasetMap = indexes.get(index.getDataverseName());
- if (datasetMap == null) {
- datasetMap = new HashMap<String, Map<String, Index>>();
- indexes.put(index.getDataverseName(), datasetMap);
- }
- Map<String, Index> indexMap = datasetMap.get(index.getDatasetName());
- if (indexMap == null) {
- indexMap = new HashMap<String, Index>();
- datasetMap.put(index.getDatasetName(), indexMap);
- }
- if (!indexMap.containsKey(index.getIndexName())) {
- return indexMap.put(index.getIndexName(), index);
- }
- return null;
- }
- }
+ public Object addIndexIfNotExists(Index index) {
+ synchronized (indexes) {
+ Map<String, Map<String, Index>> datasetMap = indexes.get(index
+ .getDataverseName());
+ if (datasetMap == null) {
+ datasetMap = new HashMap<String, Map<String, Index>>();
+ indexes.put(index.getDataverseName(), datasetMap);
+ }
+ Map<String, Index> indexMap = datasetMap
+ .get(index.getDatasetName());
+ if (indexMap == null) {
+ indexMap = new HashMap<String, Index>();
+ datasetMap.put(index.getDatasetName(), indexMap);
+ }
+ if (!indexMap.containsKey(index.getIndexName())) {
+ return indexMap.put(index.getIndexName(), index);
+ }
+ return null;
+ }
+ }
- public Object addDatatypeIfNotExists(Datatype datatype) {
- synchronized (datatypes) {
- Map<String, Datatype> m = datatypes.get(datatype.getDataverseName());
- if (m == null) {
- m = new HashMap<String, Datatype>();
- datatypes.put(datatype.getDataverseName(), m);
- }
- if (!m.containsKey(datatype.getDatatypeName())) {
- return m.put(datatype.getDatatypeName(), datatype);
- }
- return null;
- }
- }
+ public Object addDatatypeIfNotExists(Datatype datatype) {
+ synchronized (datatypes) {
+ Map<String, Datatype> m = datatypes
+ .get(datatype.getDataverseName());
+ if (m == null) {
+ m = new HashMap<String, Datatype>();
+ datatypes.put(datatype.getDataverseName(), m);
+ }
+ if (!m.containsKey(datatype.getDatatypeName())) {
+ return m.put(datatype.getDatatypeName(), datatype);
+ }
+ return null;
+ }
+ }
- public Object addNodeGroupIfNotExists(NodeGroup nodeGroup) {
- synchronized (nodeGroups) {
- if (!nodeGroups.containsKey(nodeGroup.getNodeGroupName())) {
- return nodeGroups.put(nodeGroup.getNodeGroupName(), nodeGroup);
- }
- return null;
- }
- }
+ public Object addNodeGroupIfNotExists(NodeGroup nodeGroup) {
+ synchronized (nodeGroups) {
+ if (!nodeGroups.containsKey(nodeGroup.getNodeGroupName())) {
+ return nodeGroups.put(nodeGroup.getNodeGroupName(), nodeGroup);
+ }
+ return null;
+ }
+ }
- public Object dropDataverse(Dataverse dataverse) {
- synchronized (dataverses) {
- synchronized (datasets) {
- synchronized (indexes) {
- synchronized (datatypes) {
- synchronized (functions) {
- datasets.remove(dataverse.getDataverseName());
- indexes.remove(dataverse.getDataverseName());
- datatypes.remove(dataverse.getDataverseName());
- adapters.remove(dataverse.getDataverseName());
- List<FunctionSignature> markedFunctionsForRemoval = new ArrayList<FunctionSignature>();
- for (FunctionSignature signature : functions.keySet()) {
- if (signature.getNamespace().equals(dataverse.getDataverseName())) {
- markedFunctionsForRemoval.add(signature);
- }
- }
- for (FunctionSignature signature : markedFunctionsForRemoval) {
- functions.remove(signature);
- }
- return dataverses.remove(dataverse.getDataverseName());
- }
- }
- }
- }
- }
- }
+ public Object dropDataverse(Dataverse dataverse) {
+ synchronized (dataverses) {
+ synchronized (datasets) {
+ synchronized (indexes) {
+ synchronized (datatypes) {
+ synchronized (functions) {
+ datasets.remove(dataverse.getDataverseName());
+ indexes.remove(dataverse.getDataverseName());
+ datatypes.remove(dataverse.getDataverseName());
+ adapters.remove(dataverse.getDataverseName());
+ List<FunctionSignature> markedFunctionsForRemoval = new ArrayList<FunctionSignature>();
+ for (FunctionSignature signature : functions
+ .keySet()) {
+ if (signature.getNamespace().equals(
+ dataverse.getDataverseName())) {
+ markedFunctionsForRemoval.add(signature);
+ }
+ }
+ for (FunctionSignature signature : markedFunctionsForRemoval) {
+ functions.remove(signature);
+ }
+ return dataverses.remove(dataverse
+ .getDataverseName());
+ }
+ }
+ }
+ }
+ }
+ }
- public Object dropDataset(Dataset dataset) {
- synchronized (datasets) {
- synchronized (indexes) {
+ public Object dropDataset(Dataset dataset) {
+ synchronized (datasets) {
+ synchronized (indexes) {
- //remove the indexes of the dataset from indexes' cache
- Map<String, Map<String, Index>> datasetMap = indexes.get(dataset.getDataverseName());
- if (datasetMap != null) {
- datasetMap.remove(dataset.getDatasetName());
- }
+ // remove the indexes of the dataset from indexes' cache
+ Map<String, Map<String, Index>> datasetMap = indexes
+ .get(dataset.getDataverseName());
+ if (datasetMap != null) {
+ datasetMap.remove(dataset.getDatasetName());
+ }
- //remove the dataset from datasets' cache
- Map<String, Dataset> m = datasets.get(dataset.getDataverseName());
- if (m == null) {
- return null;
- }
- return m.remove(dataset.getDatasetName());
- }
- }
- }
-
- public Object dropIndex(Index index) {
- synchronized (indexes) {
- Map<String, Map<String, Index>> datasetMap = indexes.get(index.getDataverseName());
- if (datasetMap == null) {
- return null;
- }
-
- Map<String, Index> indexMap = datasetMap.get(index.getDatasetName());
- if (indexMap == null) {
- return null;
- }
-
- return indexMap.remove(index.getIndexName());
- }
- }
+ // remove the dataset from datasets' cache
+ Map<String, Dataset> m = datasets.get(dataset
+ .getDataverseName());
+ if (m == null) {
+ return null;
+ }
+ return m.remove(dataset.getDatasetName());
+ }
+ }
+ }
- public Object dropDatatype(Datatype datatype) {
- synchronized (datatypes) {
- Map<String, Datatype> m = datatypes.get(datatype.getDataverseName());
- if (m == null) {
- return null;
- }
- return m.remove(datatype.getDatatypeName());
- }
- }
+ public Object dropIndex(Index index) {
+ synchronized (indexes) {
+ Map<String, Map<String, Index>> datasetMap = indexes.get(index
+ .getDataverseName());
+ if (datasetMap == null) {
+ return null;
+ }
- public Object dropNodeGroup(NodeGroup nodeGroup) {
- synchronized (nodeGroups) {
- return nodeGroups.remove(nodeGroup.getNodeGroupName());
- }
- }
+ Map<String, Index> indexMap = datasetMap
+ .get(index.getDatasetName());
+ if (indexMap == null) {
+ return null;
+ }
- public Dataverse getDataverse(String dataverseName) {
- synchronized (dataverses) {
- return dataverses.get(dataverseName);
- }
- }
+ return indexMap.remove(index.getIndexName());
+ }
+ }
- public Dataset getDataset(String dataverseName, String datasetName) {
- synchronized (datasets) {
- Map<String, Dataset> m = datasets.get(dataverseName);
- if (m == null) {
- return null;
- }
- return m.get(datasetName);
- }
- }
-
- public Index getIndex(String dataverseName, String datasetName, String indexName) {
- synchronized (indexes) {
- Map<String, Map<String, Index>> datasetMap = indexes.get(dataverseName);
- if (datasetMap == null) {
- return null;
- }
- Map<String, Index> indexMap = datasetMap.get(datasetName);
- if (indexMap == null) {
- return null;
- }
- return indexMap.get(indexName);
- }
- }
+ public Object dropDatatype(Datatype datatype) {
+ synchronized (datatypes) {
+ Map<String, Datatype> m = datatypes
+ .get(datatype.getDataverseName());
+ if (m == null) {
+ return null;
+ }
+ return m.remove(datatype.getDatatypeName());
+ }
+ }
- public Datatype getDatatype(String dataverseName, String datatypeName) {
- synchronized (datatypes) {
- Map<String, Datatype> m = datatypes.get(dataverseName);
- if (m == null) {
- return null;
- }
- return m.get(datatypeName);
- }
- }
+ public Object dropNodeGroup(NodeGroup nodeGroup) {
+ synchronized (nodeGroups) {
+ return nodeGroups.remove(nodeGroup.getNodeGroupName());
+ }
+ }
- public NodeGroup getNodeGroup(String nodeGroupName) {
- synchronized (nodeGroups) {
- return nodeGroups.get(nodeGroupName);
- }
- }
+ public Dataverse getDataverse(String dataverseName) {
+ synchronized (dataverses) {
+ return dataverses.get(dataverseName);
+ }
+ }
- public Function getFunction(FunctionSignature functionSignature) {
- synchronized (functions) {
- return functions.get(functionSignature);
- }
- }
+ public Dataset getDataset(String dataverseName, String datasetName) {
+ synchronized (datasets) {
+ Map<String, Dataset> m = datasets.get(dataverseName);
+ if (m == null) {
+ return null;
+ }
+ return m.get(datasetName);
+ }
+ }
- public List<Dataset> getDataverseDatasets(String dataverseName) {
- List<Dataset> retDatasets = new ArrayList<Dataset>();
- synchronized (datasets) {
- Map<String, Dataset> m = datasets.get(dataverseName);
- if (m == null) {
- return retDatasets;
- }
- for (Map.Entry<String, Dataset> entry : m.entrySet()) {
- retDatasets.add(entry.getValue());
- }
- return retDatasets;
- }
- }
+ public Index getIndex(String dataverseName, String datasetName,
+ String indexName) {
+ synchronized (indexes) {
+ Map<String, Map<String, Index>> datasetMap = indexes
+ .get(dataverseName);
+ if (datasetMap == null) {
+ return null;
+ }
+ Map<String, Index> indexMap = datasetMap.get(datasetName);
+ if (indexMap == null) {
+ return null;
+ }
+ return indexMap.get(indexName);
+ }
+ }
- /**
- * Represents a logical operation against the metadata.
- */
- protected class MetadataLogicalOperation {
- // Entity to be added/dropped.
- public final IMetadataEntity entity;
- // True for add, false for drop.
- public final boolean isAdd;
+ public Datatype getDatatype(String dataverseName, String datatypeName) {
+ synchronized (datatypes) {
+ Map<String, Datatype> m = datatypes.get(dataverseName);
+ if (m == null) {
+ return null;
+ }
+ return m.get(datatypeName);
+ }
+ }
- public MetadataLogicalOperation(IMetadataEntity entity, boolean isAdd) {
- this.entity = entity;
- this.isAdd = isAdd;
- }
- };
+ public NodeGroup getNodeGroup(String nodeGroupName) {
+ synchronized (nodeGroups) {
+ return nodeGroups.get(nodeGroupName);
+ }
+ }
- protected void doOperation(MetadataLogicalOperation op) {
- if (op.isAdd) {
- op.entity.addToCache(this);
- } else {
- op.entity.dropFromCache(this);
- }
- }
+ public Function getFunction(FunctionSignature functionSignature) {
+ synchronized (functions) {
+ return functions.get(functionSignature);
+ }
+ }
- protected void undoOperation(MetadataLogicalOperation op) {
- if (!op.isAdd) {
- op.entity.addToCache(this);
- } else {
- op.entity.dropFromCache(this);
- }
- }
+ public List<Dataset> getDataverseDatasets(String dataverseName) {
+ List<Dataset> retDatasets = new ArrayList<Dataset>();
+ synchronized (datasets) {
+ Map<String, Dataset> m = datasets.get(dataverseName);
+ if (m == null) {
+ return retDatasets;
+ }
+ for (Map.Entry<String, Dataset> entry : m.entrySet()) {
+ retDatasets.add(entry.getValue());
+ }
+ return retDatasets;
+ }
+ }
- public Object addFunctionIfNotExists(Function function) {
- synchronized (functions) {
- FunctionSignature signature = new FunctionSignature(function.getDataverseName(), function.getName(),
- function.getArity());
- Function fun = functions.get(signature);
- if (fun == null) {
- return functions.put(signature, function);
- }
- return null;
- }
- }
+ /**
+ * Represents a logical operation against the metadata.
+ */
+ protected class MetadataLogicalOperation {
+ // Entity to be added/dropped.
+ public final IMetadataEntity entity;
+ // True for add, false for drop.
+ public final boolean isAdd;
- public Object dropFunction(Function function) {
- synchronized (functions) {
- FunctionSignature signature = new FunctionSignature(function.getDataverseName(), function.getName(),
- function.getArity());
- Function fun = functions.get(signature);
- if (fun == null) {
- return null;
- }
- return functions.remove(signature);
- }
- }
+ public MetadataLogicalOperation(IMetadataEntity entity, boolean isAdd) {
+ this.entity = entity;
+ this.isAdd = isAdd;
+ }
+ };
- public Object addAdapterIfNotExists(DatasourceAdapter adapter) {
- synchronized (adapters) {
- DatasourceAdapter adapterObject = adapters.get(adapter.getAdapterIdentifier().getNamespace()).get(
- adapter.getAdapterIdentifier().getAdapterName());
- if (adapterObject != null) {
- Map<String, DatasourceAdapter> adaptersInDataverse = adapters.get(adapter.getAdapterIdentifier()
- .getNamespace());
- if (adaptersInDataverse == null) {
- adaptersInDataverse = new HashMap<String, DatasourceAdapter>();
- adapters.put(adapter.getAdapterIdentifier().getNamespace(), adaptersInDataverse);
- }
- return adaptersInDataverse.put(adapter.getAdapterIdentifier().getAdapterName(), adapter);
- }
- return null;
- }
- }
+ protected void doOperation(MetadataLogicalOperation op) {
+ if (op.isAdd) {
+ op.entity.addToCache(this);
+ } else {
+ op.entity.dropFromCache(this);
+ }
+ }
- public Object dropAdapter(DatasourceAdapter adapter) {
- synchronized (adapters) {
- Map<String, DatasourceAdapter> adaptersInDataverse = adapters.get(adapter.getAdapterIdentifier()
- .getNamespace());
- if (adaptersInDataverse != null) {
- return adaptersInDataverse.remove(adapter.getAdapterIdentifier().getAdapterName());
- }
- return null;
- }
- }
+ protected void undoOperation(MetadataLogicalOperation op) {
+ if (!op.isAdd) {
+ op.entity.addToCache(this);
+ } else {
+ op.entity.dropFromCache(this);
+ }
+ }
+
+ public Object addFunctionIfNotExists(Function function) {
+ synchronized (functions) {
+ FunctionSignature signature = new FunctionSignature(
+ function.getDataverseName(), function.getName(),
+ function.getArity());
+ Function fun = functions.get(signature);
+ if (fun == null) {
+ return functions.put(signature, function);
+ }
+ return null;
+ }
+ }
+
+ public Object dropFunction(Function function) {
+ synchronized (functions) {
+ FunctionSignature signature = new FunctionSignature(
+ function.getDataverseName(), function.getName(),
+ function.getArity());
+ Function fun = functions.get(signature);
+ if (fun == null) {
+ return null;
+ }
+ return functions.remove(signature);
+ }
+ }
+
+ public Object addAdapterIfNotExists(DatasourceAdapter adapter) {
+ synchronized (adapters) {
+ boolean needToAddd = adapters.get(adapter.getAdapterIdentifier()) == null;
+ if (needToAddd) {
+ return adapters.put(adapter.getAdapterIdentifier(), adapter);
+ }
+ return null;
+ }
+ }
+
+ public Object dropAdapter(DatasourceAdapter adapter) {
+ synchronized (adapters) {
+ if (adapters.get(adapter.getAdapterIdentifier()) != null) {
+ return adapters.remove(adapter.getAdapterIdentifier());
+ }
+ return null;
+ }
+ }
+
+ public Object addLibraryIfNotExists(Library library) {
+ synchronized (libraries) {
+ Map<String, Library> libsInDataverse = libraries.get(library
+ .getDataverseName());
+ boolean needToAddd = (libsInDataverse == null || libsInDataverse
+ .get(library.getName()) != null);
+ if (needToAddd) {
+ if (libsInDataverse == null) {
+ libsInDataverse = new HashMap<String, Library>();
+ libraries.put(library.getDataverseName(), libsInDataverse);
+ }
+ return libsInDataverse.put(library.getDataverseName(), library);
+ }
+ return null;
+ }
+ }
+
+ public Object dropLibrary(Library library) {
+ synchronized (libraries) {
+ Map<String, Library> librariesInDataverse = libraries.get(library
+ .getDataverseName());
+ if (librariesInDataverse != null) {
+ return librariesInDataverse.remove(library.getName());
+ }
+ return null;
+ }
+ }
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataManager.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataManager.java
index ad106fe..a573ed1 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataManager.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataManager.java
@@ -31,6 +31,7 @@
import edu.uci.ics.asterix.metadata.entities.Dataverse;
import edu.uci.ics.asterix.metadata.entities.Function;
import edu.uci.ics.asterix.metadata.entities.Index;
+import edu.uci.ics.asterix.metadata.entities.Library;
import edu.uci.ics.asterix.metadata.entities.Node;
import edu.uci.ics.asterix.metadata.entities.NodeGroup;
import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
@@ -43,563 +44,642 @@
* received from the metadata node, to avoid contacting the metadata node
* repeatedly. We assume that this metadata manager is the only metadata manager
* in an Asterix cluster. Therefore, no separate cache-invalidation mechanism is
- * needed at this point.
- * Assumptions/Limitations:
- * The metadata subsystem is started during NC Bootstrap start, i.e., when
- * Asterix is deployed.
- * The metadata subsystem is destroyed in NC Bootstrap end, i.e., when Asterix
- * is undeployed.
- * The metadata subsystem consists of the MetadataManager and the MatadataNode.
- * The MetadataManager provides users access to the metadata.
- * The MetadataNode implements direct access to the storage layer on behalf of
- * the MetadataManager, and translates the binary representation of ADM into
- * Java objects for consumption by the MetadataManager's users.
- * There is exactly one instance of the MetadataManager and of the MetadataNode
- * in the cluster, which may or may not be co-located on the same machine (or in
- * the same JVM).
- * The MetadataManager exists in the same JVM as its user's (e.g., the query
- * compiler).
- * The MetadataNode exists in the same JVM as it's transactional components
- * (LockManager, LogManager, etc.)
- * Users shall access the metadata only through the MetadataManager, and never
- * via the MetadataNode directly.
+ * needed at this point. Assumptions/Limitations: The metadata subsystem is
+ * started during NC Bootstrap start, i.e., when Asterix is deployed. The
+ * metadata subsystem is destroyed in NC Bootstrap end, i.e., when Asterix is
+ * undeployed. The metadata subsystem consists of the MetadataManager and the
+ * MatadataNode. The MetadataManager provides users access to the metadata. The
+ * MetadataNode implements direct access to the storage layer on behalf of the
+ * MetadataManager, and translates the binary representation of ADM into Java
+ * objects for consumption by the MetadataManager's users. There is exactly one
+ * instance of the MetadataManager and of the MetadataNode in the cluster, which
+ * may or may not be co-located on the same machine (or in the same JVM). The
+ * MetadataManager exists in the same JVM as its user's (e.g., the query
+ * compiler). The MetadataNode exists in the same JVM as it's transactional
+ * components (LockManager, LogManager, etc.) Users shall access the metadata
+ * only through the MetadataManager, and never via the MetadataNode directly.
* Multiple threads may issue requests to the MetadataManager concurrently. For
* the sake of accessing metadata, we assume a transaction consists of one
- * thread.
- * Users are responsible for locking the metadata (using the MetadataManager
- * API) before issuing requests.
- * The MetadataNode is responsible for acquiring finer-grained locks on behalf
- * of requests from the MetadataManager. Currently, locks are acquired per
- * BTree, since the BTree does not acquire even finer-grained locks yet
- * internally.
- * The metadata can be queried with AQL DML like any other dataset, but can only
- * be changed with AQL DDL.
- * The transaction ids for metadata transactions must be unique across the
+ * thread. Users are responsible for locking the metadata (using the
+ * MetadataManager API) before issuing requests. The MetadataNode is responsible
+ * for acquiring finer-grained locks on behalf of requests from the
+ * MetadataManager. Currently, locks are acquired per BTree, since the BTree
+ * does not acquire even finer-grained locks yet internally. The metadata can be
+ * queried with AQL DML like any other dataset, but can only be changed with AQL
+ * DDL. The transaction ids for metadata transactions must be unique across the
* cluster, i.e., metadata transaction ids shall never "accidentally" overlap
* with transaction ids of regular jobs or other metadata transactions.
*/
public class MetadataManager implements IMetadataManager {
- // Set in init().
- public static MetadataManager INSTANCE;
- private final MetadataCache cache = new MetadataCache();
- private IAsterixStateProxy proxy;
- private IMetadataNode metadataNode;
- private final ReadWriteLock metadataLatch;
+ // Set in init().
+ public static MetadataManager INSTANCE;
+ private final MetadataCache cache = new MetadataCache();
+ private IAsterixStateProxy proxy;
+ private IMetadataNode metadataNode;
+ private final ReadWriteLock metadataLatch;
- public MetadataManager(IAsterixStateProxy proxy) {
- if (proxy == null) {
- throw new Error("Null proxy given to MetadataManager.");
- }
- this.proxy = proxy;
- this.metadataNode = null;
- this.metadataLatch = new ReentrantReadWriteLock(true);
- }
+ public MetadataManager(IAsterixStateProxy proxy) {
+ if (proxy == null) {
+ throw new Error("Null proxy given to MetadataManager.");
+ }
+ this.proxy = proxy;
+ this.metadataNode = null;
+ this.metadataLatch = new ReentrantReadWriteLock(true);
+ }
- @Override
- public void init() throws RemoteException {
- // Could be synchronized on any object. Arbitrarily chose proxy.
- synchronized (proxy) {
- if (metadataNode != null) {
- return;
- }
- metadataNode = proxy.getMetadataNode();
- if (metadataNode == null) {
- throw new Error("Failed to get the MetadataNode.\n" + "The MetadataNode was configured to run on NC: "
- + proxy.getAsterixProperties().getMetadataNodeName());
- }
- }
- }
+ @Override
+ public void init() throws RemoteException {
+ // Could be synchronized on any object. Arbitrarily chose proxy.
+ synchronized (proxy) {
+ if (metadataNode != null) {
+ return;
+ }
+ metadataNode = proxy.getMetadataNode();
+ if (metadataNode == null) {
+ throw new Error("Failed to get the MetadataNode.\n"
+ + "The MetadataNode was configured to run on NC: "
+ + proxy.getAsterixProperties().getMetadataNodeName());
+ }
+ }
+ }
- @Override
- public MetadataTransactionContext beginTransaction() throws RemoteException, ACIDException {
- JobId jobId = JobIdFactory.generateJobId();
- metadataNode.beginTransaction(jobId);
- return new MetadataTransactionContext(jobId);
- }
+ @Override
+ public MetadataTransactionContext beginTransaction()
+ throws RemoteException, ACIDException {
+ JobId jobId = JobIdFactory.generateJobId();
+ metadataNode.beginTransaction(jobId);
+ return new MetadataTransactionContext(jobId);
+ }
- @Override
- public void commitTransaction(MetadataTransactionContext ctx) throws RemoteException, ACIDException {
- metadataNode.commitTransaction(ctx.getJobId());
- cache.commit(ctx);
- }
+ @Override
+ public void commitTransaction(MetadataTransactionContext ctx)
+ throws RemoteException, ACIDException {
+ metadataNode.commitTransaction(ctx.getJobId());
+ cache.commit(ctx);
+ }
- @Override
- public void abortTransaction(MetadataTransactionContext ctx) throws RemoteException, ACIDException {
- metadataNode.abortTransaction(ctx.getJobId());
- }
+ @Override
+ public void abortTransaction(MetadataTransactionContext ctx)
+ throws RemoteException, ACIDException {
+ metadataNode.abortTransaction(ctx.getJobId());
+ }
- @Override
- public void lock(MetadataTransactionContext ctx, byte lockMode) throws RemoteException, ACIDException {
- metadataNode.lock(ctx.getJobId(), lockMode);
- }
+ @Override
+ public void lock(MetadataTransactionContext ctx, byte lockMode)
+ throws RemoteException, ACIDException {
+ metadataNode.lock(ctx.getJobId(), lockMode);
+ }
- @Override
- public void unlock(MetadataTransactionContext ctx) throws RemoteException, ACIDException {
- metadataNode.unlock(ctx.getJobId());
- }
+ @Override
+ public void unlock(MetadataTransactionContext ctx) throws RemoteException,
+ ACIDException {
+ metadataNode.unlock(ctx.getJobId());
+ }
- @Override
- public void addDataverse(MetadataTransactionContext ctx, Dataverse dataverse) throws MetadataException {
- try {
- metadataNode.addDataverse(ctx.getJobId(), dataverse);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- ctx.addDataverse(dataverse);
- }
+ @Override
+ public void addDataverse(MetadataTransactionContext ctx, Dataverse dataverse)
+ throws MetadataException {
+ try {
+ metadataNode.addDataverse(ctx.getJobId(), dataverse);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.addDataverse(dataverse);
+ }
- @Override
- public void dropDataverse(MetadataTransactionContext ctx, String dataverseName) throws MetadataException {
- try {
- metadataNode.dropDataverse(ctx.getJobId(), dataverseName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- ctx.dropDataverse(dataverseName);
- }
+ @Override
+ public void dropDataverse(MetadataTransactionContext ctx,
+ String dataverseName) throws MetadataException {
+ try {
+ metadataNode.dropDataverse(ctx.getJobId(), dataverseName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.dropDataverse(dataverseName);
+ }
- @Override
- public List<Dataverse> getDataverses(MetadataTransactionContext ctx) throws MetadataException {
- try {
- return metadataNode.getDataverses(ctx.getJobId());
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- }
+ @Override
+ public List<Dataverse> getDataverses(MetadataTransactionContext ctx)
+ throws MetadataException {
+ try {
+ return metadataNode.getDataverses(ctx.getJobId());
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ }
- @Override
- public Dataverse getDataverse(MetadataTransactionContext ctx, String dataverseName) throws MetadataException {
- // First look in the context to see if this transaction created the
- // requested dataverse itself (but the dataverse is still uncommitted).
- Dataverse dataverse = ctx.getDataverse(dataverseName);
- if (dataverse != null) {
- // Don't add this dataverse to the cache, since it is still
- // uncommitted.
- return dataverse;
- }
- if (ctx.dataverseIsDropped(dataverseName)) {
- // Dataverse has been dropped by this transaction but could still be
- // in the cache.
- return null;
- }
- dataverse = cache.getDataverse(dataverseName);
- if (dataverse != null) {
- // Dataverse is already in the cache, don't add it again.
- return dataverse;
- }
- try {
- dataverse = metadataNode.getDataverse(ctx.getJobId(), dataverseName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- // We fetched the dataverse from the MetadataNode. Add it to the cache
- // when this transaction commits.
- if (dataverse != null) {
- ctx.addDataverse(dataverse);
- }
- return dataverse;
- }
+ @Override
+ public Dataverse getDataverse(MetadataTransactionContext ctx,
+ String dataverseName) throws MetadataException {
+ // First look in the context to see if this transaction created the
+ // requested dataverse itself (but the dataverse is still uncommitted).
+ Dataverse dataverse = ctx.getDataverse(dataverseName);
+ if (dataverse != null) {
+ // Don't add this dataverse to the cache, since it is still
+ // uncommitted.
+ return dataverse;
+ }
+ if (ctx.dataverseIsDropped(dataverseName)) {
+ // Dataverse has been dropped by this transaction but could still be
+ // in the cache.
+ return null;
+ }
+ dataverse = cache.getDataverse(dataverseName);
+ if (dataverse != null) {
+ // Dataverse is already in the cache, don't add it again.
+ return dataverse;
+ }
+ try {
+ dataverse = metadataNode
+ .getDataverse(ctx.getJobId(), dataverseName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ // We fetched the dataverse from the MetadataNode. Add it to the cache
+ // when this transaction commits.
+ if (dataverse != null) {
+ ctx.addDataverse(dataverse);
+ }
+ return dataverse;
+ }
- @Override
- public List<Dataset> getDataverseDatasets(MetadataTransactionContext ctx, String dataverseName)
- throws MetadataException {
- List<Dataset> dataverseDatasets;
- try {
- // Assuming that the transaction can read its own writes on the
- // metadata node.
- dataverseDatasets = metadataNode.getDataverseDatasets(ctx.getJobId(), dataverseName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- // Don't update the cache to avoid checking against the transaction's
- // uncommitted datasets.
- return dataverseDatasets;
- }
+ @Override
+ public List<Dataset> getDataverseDatasets(MetadataTransactionContext ctx,
+ String dataverseName) throws MetadataException {
+ List<Dataset> dataverseDatasets;
+ try {
+ // Assuming that the transaction can read its own writes on the
+ // metadata node.
+ dataverseDatasets = metadataNode.getDataverseDatasets(
+ ctx.getJobId(), dataverseName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ // Don't update the cache to avoid checking against the transaction's
+ // uncommitted datasets.
+ return dataverseDatasets;
+ }
- @Override
- public void addDataset(MetadataTransactionContext ctx, Dataset dataset) throws MetadataException {
- // add dataset into metadataNode
- try {
- metadataNode.addDataset(ctx.getJobId(), dataset);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
+ @Override
+ public void addDataset(MetadataTransactionContext ctx, Dataset dataset)
+ throws MetadataException {
+ // add dataset into metadataNode
+ try {
+ metadataNode.addDataset(ctx.getJobId(), dataset);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
- // reflect the dataset into the cache
- ctx.addDataset(dataset);
- }
+ // reflect the dataset into the cache
+ ctx.addDataset(dataset);
+ }
- @Override
- public void dropDataset(MetadataTransactionContext ctx, String dataverseName, String datasetName)
- throws MetadataException {
- try {
- metadataNode.dropDataset(ctx.getJobId(), dataverseName, datasetName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- ctx.dropDataset(dataverseName, datasetName);
- }
+ @Override
+ public void dropDataset(MetadataTransactionContext ctx,
+ String dataverseName, String datasetName) throws MetadataException {
+ try {
+ metadataNode
+ .dropDataset(ctx.getJobId(), dataverseName, datasetName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.dropDataset(dataverseName, datasetName);
+ }
- @Override
- public Dataset getDataset(MetadataTransactionContext ctx, String dataverseName, String datasetName)
- throws MetadataException {
+ @Override
+ public Dataset getDataset(MetadataTransactionContext ctx,
+ String dataverseName, String datasetName) throws MetadataException {
- // First look in the context to see if this transaction created the
- // requested dataset itself (but the dataset is still uncommitted).
- Dataset dataset = ctx.getDataset(dataverseName, datasetName);
- if (dataset != null) {
- // Don't add this dataverse to the cache, since it is still
- // uncommitted.
- return dataset;
- }
- if (ctx.datasetIsDropped(dataverseName, datasetName)) {
- // Dataset has been dropped by this transaction but could still be
- // in the cache.
- return null;
- }
+ // First look in the context to see if this transaction created the
+ // requested dataset itself (but the dataset is still uncommitted).
+ Dataset dataset = ctx.getDataset(dataverseName, datasetName);
+ if (dataset != null) {
+ // Don't add this dataverse to the cache, since it is still
+ // uncommitted.
+ return dataset;
+ }
+ if (ctx.datasetIsDropped(dataverseName, datasetName)) {
+ // Dataset has been dropped by this transaction but could still be
+ // in the cache.
+ return null;
+ }
- dataset = cache.getDataset(dataverseName, datasetName);
- if (dataset != null) {
- // Dataset is already in the cache, don't add it again.
- return dataset;
- }
- try {
- dataset = metadataNode.getDataset(ctx.getJobId(), dataverseName, datasetName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- // We fetched the dataset from the MetadataNode. Add it to the cache
- // when this transaction commits.
- if (dataset != null) {
- ctx.addDataset(dataset);
- }
- return dataset;
- }
+ dataset = cache.getDataset(dataverseName, datasetName);
+ if (dataset != null) {
+ // Dataset is already in the cache, don't add it again.
+ return dataset;
+ }
+ try {
+ dataset = metadataNode.getDataset(ctx.getJobId(), dataverseName,
+ datasetName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ // We fetched the dataset from the MetadataNode. Add it to the cache
+ // when this transaction commits.
+ if (dataset != null) {
+ ctx.addDataset(dataset);
+ }
+ return dataset;
+ }
- @Override
- public List<Index> getDatasetIndexes(MetadataTransactionContext ctx, String dataverseName, String datasetName)
- throws MetadataException {
- List<Index> datsetIndexes;
- try {
- datsetIndexes = metadataNode.getDatasetIndexes(ctx.getJobId(), dataverseName, datasetName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- return datsetIndexes;
- }
+ @Override
+ public List<Index> getDatasetIndexes(MetadataTransactionContext ctx,
+ String dataverseName, String datasetName) throws MetadataException {
+ List<Index> datsetIndexes;
+ try {
+ datsetIndexes = metadataNode.getDatasetIndexes(ctx.getJobId(),
+ dataverseName, datasetName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ return datsetIndexes;
+ }
- @Override
- public void addDatatype(MetadataTransactionContext ctx, Datatype datatype) throws MetadataException {
- try {
- metadataNode.addDatatype(ctx.getJobId(), datatype);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- ctx.addDatatype(datatype);
- }
+ @Override
+ public void addDatatype(MetadataTransactionContext ctx, Datatype datatype)
+ throws MetadataException {
+ try {
+ metadataNode.addDatatype(ctx.getJobId(), datatype);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.addDatatype(datatype);
+ }
- @Override
- public void dropDatatype(MetadataTransactionContext ctx, String dataverseName, String datatypeName)
- throws MetadataException {
- try {
- metadataNode.dropDatatype(ctx.getJobId(), dataverseName, datatypeName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- ctx.dropDataDatatype(dataverseName, datatypeName);
- }
+ @Override
+ public void dropDatatype(MetadataTransactionContext ctx,
+ String dataverseName, String datatypeName) throws MetadataException {
+ try {
+ metadataNode.dropDatatype(ctx.getJobId(), dataverseName,
+ datatypeName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.dropDataDatatype(dataverseName, datatypeName);
+ }
- @Override
- public Datatype getDatatype(MetadataTransactionContext ctx, String dataverseName, String datatypeName)
- throws MetadataException {
- // First look in the context to see if this transaction created the
- // requested datatype itself (but the datatype is still uncommitted).
- Datatype datatype = ctx.getDatatype(dataverseName, datatypeName);
- if (datatype != null) {
- // Don't add this dataverse to the cache, since it is still
- // uncommitted.
- return datatype;
- }
- if (ctx.datatypeIsDropped(dataverseName, datatypeName)) {
- // Datatype has been dropped by this transaction but could still be
- // in the cache.
- return null;
- }
+ @Override
+ public Datatype getDatatype(MetadataTransactionContext ctx,
+ String dataverseName, String datatypeName) throws MetadataException {
+ // First look in the context to see if this transaction created the
+ // requested datatype itself (but the datatype is still uncommitted).
+ Datatype datatype = ctx.getDatatype(dataverseName, datatypeName);
+ if (datatype != null) {
+ // Don't add this dataverse to the cache, since it is still
+ // uncommitted.
+ return datatype;
+ }
+ if (ctx.datatypeIsDropped(dataverseName, datatypeName)) {
+ // Datatype has been dropped by this transaction but could still be
+ // in the cache.
+ return null;
+ }
- datatype = cache.getDatatype(dataverseName, datatypeName);
- if (datatype != null) {
- // Datatype is already in the cache, don't add it again.
- return datatype;
- }
- try {
- datatype = metadataNode.getDatatype(ctx.getJobId(), dataverseName, datatypeName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- // We fetched the datatype from the MetadataNode. Add it to the cache
- // when this transaction commits.
- if (datatype != null) {
- ctx.addDatatype(datatype);
- }
- return datatype;
- }
+ datatype = cache.getDatatype(dataverseName, datatypeName);
+ if (datatype != null) {
+ // Datatype is already in the cache, don't add it again.
+ return datatype;
+ }
+ try {
+ datatype = metadataNode.getDatatype(ctx.getJobId(), dataverseName,
+ datatypeName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ // We fetched the datatype from the MetadataNode. Add it to the cache
+ // when this transaction commits.
+ if (datatype != null) {
+ ctx.addDatatype(datatype);
+ }
+ return datatype;
+ }
- @Override
- public void addIndex(MetadataTransactionContext ctx, Index index) throws MetadataException {
- try {
- metadataNode.addIndex(ctx.getJobId(), index);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- ctx.addIndex(index);
- }
+ @Override
+ public void addIndex(MetadataTransactionContext ctx, Index index)
+ throws MetadataException {
+ try {
+ metadataNode.addIndex(ctx.getJobId(), index);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.addIndex(index);
+ }
- @Override
- public void addAdapter(MetadataTransactionContext mdTxnCtx, DatasourceAdapter adapter) throws MetadataException {
- try {
- metadataNode.addAdapter(mdTxnCtx.getJobId(), adapter);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- mdTxnCtx.addAdapter(adapter);
+ @Override
+ public void addAdapter(MetadataTransactionContext mdTxnCtx,
+ DatasourceAdapter adapter) throws MetadataException {
+ try {
+ metadataNode.addAdapter(mdTxnCtx.getJobId(), adapter);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ mdTxnCtx.addAdapter(adapter);
- }
+ }
- @Override
- public void dropIndex(MetadataTransactionContext ctx, String dataverseName, String datasetName, String indexName)
- throws MetadataException {
- try {
- metadataNode.dropIndex(ctx.getJobId(), dataverseName, datasetName, indexName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- ctx.dropIndex(dataverseName, datasetName, indexName);
- }
+ @Override
+ public void dropIndex(MetadataTransactionContext ctx, String dataverseName,
+ String datasetName, String indexName) throws MetadataException {
+ try {
+ metadataNode.dropIndex(ctx.getJobId(), dataverseName, datasetName,
+ indexName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.dropIndex(dataverseName, datasetName, indexName);
+ }
- @Override
- public Index getIndex(MetadataTransactionContext ctx, String dataverseName, String datasetName, String indexName)
- throws MetadataException {
+ @Override
+ public Index getIndex(MetadataTransactionContext ctx, String dataverseName,
+ String datasetName, String indexName) throws MetadataException {
- // First look in the context to see if this transaction created the
- // requested index itself (but the index is still uncommitted).
- Index index = ctx.getIndex(dataverseName, datasetName, indexName);
- if (index != null) {
- // Don't add this index to the cache, since it is still
- // uncommitted.
- return index;
- }
+ // First look in the context to see if this transaction created the
+ // requested index itself (but the index is still uncommitted).
+ Index index = ctx.getIndex(dataverseName, datasetName, indexName);
+ if (index != null) {
+ // Don't add this index to the cache, since it is still
+ // uncommitted.
+ return index;
+ }
- if (ctx.indexIsDropped(dataverseName, datasetName, indexName)) {
- // Index has been dropped by this transaction but could still be
- // in the cache.
- return null;
- }
+ if (ctx.indexIsDropped(dataverseName, datasetName, indexName)) {
+ // Index has been dropped by this transaction but could still be
+ // in the cache.
+ return null;
+ }
- index = cache.getIndex(dataverseName, datasetName, indexName);
- if (index != null) {
- // Index is already in the cache, don't add it again.
- return index;
- }
- try {
- index = metadataNode.getIndex(ctx.getJobId(), dataverseName, datasetName, indexName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- // We fetched the index from the MetadataNode. Add it to the cache
- // when this transaction commits.
- if (index != null) {
- ctx.addIndex(index);
- }
- return index;
- }
+ index = cache.getIndex(dataverseName, datasetName, indexName);
+ if (index != null) {
+ // Index is already in the cache, don't add it again.
+ return index;
+ }
+ try {
+ index = metadataNode.getIndex(ctx.getJobId(), dataverseName,
+ datasetName, indexName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ // We fetched the index from the MetadataNode. Add it to the cache
+ // when this transaction commits.
+ if (index != null) {
+ ctx.addIndex(index);
+ }
+ return index;
+ }
- @Override
- public void addNode(MetadataTransactionContext ctx, Node node) throws MetadataException {
- try {
- metadataNode.addNode(ctx.getJobId(), node);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- }
+ @Override
+ public void addNode(MetadataTransactionContext ctx, Node node)
+ throws MetadataException {
+ try {
+ metadataNode.addNode(ctx.getJobId(), node);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ }
- @Override
- public void addNodegroup(MetadataTransactionContext ctx, NodeGroup nodeGroup) throws MetadataException {
- try {
- metadataNode.addNodeGroup(ctx.getJobId(), nodeGroup);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- ctx.addNogeGroup(nodeGroup);
- }
+ @Override
+ public void addNodegroup(MetadataTransactionContext ctx, NodeGroup nodeGroup)
+ throws MetadataException {
+ try {
+ metadataNode.addNodeGroup(ctx.getJobId(), nodeGroup);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.addNogeGroup(nodeGroup);
+ }
- @Override
- public void dropNodegroup(MetadataTransactionContext ctx, String nodeGroupName) throws MetadataException {
- try {
- metadataNode.dropNodegroup(ctx.getJobId(), nodeGroupName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- ctx.dropNodeGroup(nodeGroupName);
- }
+ @Override
+ public void dropNodegroup(MetadataTransactionContext ctx,
+ String nodeGroupName) throws MetadataException {
+ try {
+ metadataNode.dropNodegroup(ctx.getJobId(), nodeGroupName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.dropNodeGroup(nodeGroupName);
+ }
- @Override
- public NodeGroup getNodegroup(MetadataTransactionContext ctx, String nodeGroupName) throws MetadataException {
- // First look in the context to see if this transaction created the
- // requested dataverse itself (but the dataverse is still uncommitted).
- NodeGroup nodeGroup = ctx.getNodeGroup(nodeGroupName);
- if (nodeGroup != null) {
- // Don't add this dataverse to the cache, since it is still
- // uncommitted.
- return nodeGroup;
- }
- if (ctx.nodeGroupIsDropped(nodeGroupName)) {
- // NodeGroup has been dropped by this transaction but could still be
- // in the cache.
- return null;
- }
- nodeGroup = cache.getNodeGroup(nodeGroupName);
- if (nodeGroup != null) {
- // NodeGroup is already in the cache, don't add it again.
- return nodeGroup;
- }
- try {
- nodeGroup = metadataNode.getNodeGroup(ctx.getJobId(), nodeGroupName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- // We fetched the nodeGroup from the MetadataNode. Add it to the cache
- // when this transaction commits.
- if (nodeGroup != null) {
- ctx.addNogeGroup(nodeGroup);
- }
- return nodeGroup;
- }
+ @Override
+ public NodeGroup getNodegroup(MetadataTransactionContext ctx,
+ String nodeGroupName) throws MetadataException {
+ // First look in the context to see if this transaction created the
+ // requested dataverse itself (but the dataverse is still uncommitted).
+ NodeGroup nodeGroup = ctx.getNodeGroup(nodeGroupName);
+ if (nodeGroup != null) {
+ // Don't add this dataverse to the cache, since it is still
+ // uncommitted.
+ return nodeGroup;
+ }
+ if (ctx.nodeGroupIsDropped(nodeGroupName)) {
+ // NodeGroup has been dropped by this transaction but could still be
+ // in the cache.
+ return null;
+ }
+ nodeGroup = cache.getNodeGroup(nodeGroupName);
+ if (nodeGroup != null) {
+ // NodeGroup is already in the cache, don't add it again.
+ return nodeGroup;
+ }
+ try {
+ nodeGroup = metadataNode
+ .getNodeGroup(ctx.getJobId(), nodeGroupName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ // We fetched the nodeGroup from the MetadataNode. Add it to the cache
+ // when this transaction commits.
+ if (nodeGroup != null) {
+ ctx.addNogeGroup(nodeGroup);
+ }
+ return nodeGroup;
+ }
- @Override
- public void addFunction(MetadataTransactionContext mdTxnCtx, Function function) throws MetadataException {
- try {
- metadataNode.addFunction(mdTxnCtx.getJobId(), function);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- mdTxnCtx.addFunction(function);
- }
+ @Override
+ public void addFunction(MetadataTransactionContext mdTxnCtx,
+ Function function) throws MetadataException {
+ try {
+ metadataNode.addFunction(mdTxnCtx.getJobId(), function);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ mdTxnCtx.addFunction(function);
+ }
- @Override
- public void dropFunction(MetadataTransactionContext ctx, FunctionSignature functionSignature)
- throws MetadataException {
- try {
- metadataNode.dropFunction(ctx.getJobId(), functionSignature);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- ctx.dropFunction(functionSignature);
- }
+ @Override
+ public void dropFunction(MetadataTransactionContext ctx,
+ FunctionSignature functionSignature) throws MetadataException {
+ try {
+ metadataNode.dropFunction(ctx.getJobId(), functionSignature);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.dropFunction(functionSignature);
+ }
- @Override
- public Function getFunction(MetadataTransactionContext ctx, FunctionSignature functionSignature)
- throws MetadataException {
- // First look in the context to see if this transaction created the
- // requested dataset itself (but the dataset is still uncommitted).
- Function function = ctx.getFunction(functionSignature);
- if (function != null) {
- // Don't add this dataverse to the cache, since it is still
- // uncommitted.
- return function;
- }
- if (ctx.functionIsDropped(functionSignature)) {
- // Function has been dropped by this transaction but could still be
- // in the cache.
- return null;
- }
- if (ctx.getDataverse(functionSignature.getNamespace()) != null) {
- // This transaction has dropped and subsequently created the same
- // dataverse.
- return null;
- }
- function = cache.getFunction(functionSignature);
- if (function != null) {
- // Function is already in the cache, don't add it again.
- return function;
- }
- try {
- function = metadataNode.getFunction(ctx.getJobId(), functionSignature);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- // We fetched the function from the MetadataNode. Add it to the cache
- // when this transaction commits.
- if (function != null) {
- ctx.addFunction(function);
- }
- return function;
+ @Override
+ public Function getFunction(MetadataTransactionContext ctx,
+ FunctionSignature functionSignature) throws MetadataException {
+ // First look in the context to see if this transaction created the
+ // requested dataset itself (but the dataset is still uncommitted).
+ Function function = ctx.getFunction(functionSignature);
+ if (function != null) {
+ // Don't add this dataverse to the cache, since it is still
+ // uncommitted.
+ return function;
+ }
+ if (ctx.functionIsDropped(functionSignature)) {
+ // Function has been dropped by this transaction but could still be
+ // in the cache.
+ return null;
+ }
+ if (ctx.getDataverse(functionSignature.getNamespace()) != null) {
+ // This transaction has dropped and subsequently created the same
+ // dataverse.
+ return null;
+ }
+ function = cache.getFunction(functionSignature);
+ if (function != null) {
+ // Function is already in the cache, don't add it again.
+ return function;
+ }
+ try {
+ function = metadataNode.getFunction(ctx.getJobId(),
+ functionSignature);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ // We fetched the function from the MetadataNode. Add it to the cache
+ // when this transaction commits.
+ if (function != null) {
+ ctx.addFunction(function);
+ }
+ return function;
- }
+ }
- @Override
- public void initializeDatasetIdFactory(MetadataTransactionContext ctx) throws MetadataException {
- try {
- metadataNode.initializeDatasetIdFactory(ctx.getJobId());
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- }
+ @Override
+ public void initializeDatasetIdFactory(MetadataTransactionContext ctx)
+ throws MetadataException {
+ try {
+ metadataNode.initializeDatasetIdFactory(ctx.getJobId());
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ }
- @Override
- public List<Function> getDataverseFunctions(MetadataTransactionContext ctx, String dataverseName)
- throws MetadataException {
- List<Function> dataverseFunctions;
- try {
- // Assuming that the transaction can read its own writes on the
- // metadata node.
- dataverseFunctions = metadataNode.getDataverseFunctions(ctx.getJobId(), dataverseName);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- // Don't update the cache to avoid checking against the transaction's
- // uncommitted functions.
- return dataverseFunctions;
- }
+ @Override
+ public List<Function> getDataverseFunctions(MetadataTransactionContext ctx,
+ String dataverseName) throws MetadataException {
+ List<Function> dataverseFunctions;
+ try {
+ // Assuming that the transaction can read its own writes on the
+ // metadata node.
+ dataverseFunctions = metadataNode.getDataverseFunctions(
+ ctx.getJobId(), dataverseName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ // Don't update the cache to avoid checking against the transaction's
+ // uncommitted functions.
+ return dataverseFunctions;
+ }
- @Override
- public void dropAdapter(MetadataTransactionContext ctx, String dataverseName, String name) throws MetadataException {
- try {
- metadataNode.dropAdapter(ctx.getJobId(), dataverseName, name);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- }
+ @Override
+ public void dropAdapter(MetadataTransactionContext ctx,
+ String dataverseName, String name) throws MetadataException {
+ try {
+ metadataNode.dropAdapter(ctx.getJobId(), dataverseName, name);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ }
- @Override
- public DatasourceAdapter getAdapter(MetadataTransactionContext ctx, String dataverseName, String name)
- throws MetadataException {
- DatasourceAdapter adapter = null;
- try {
- adapter = metadataNode.getAdapter(ctx.getJobId(), dataverseName, name);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
- return adapter;
- }
+ @Override
+ public DatasourceAdapter getAdapter(MetadataTransactionContext ctx,
+ String dataverseName, String name) throws MetadataException {
+ DatasourceAdapter adapter = null;
+ try {
+ adapter = metadataNode.getAdapter(ctx.getJobId(), dataverseName,
+ name);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ return adapter;
+ }
- @Override
- public void acquireWriteLatch() {
- metadataLatch.writeLock().lock();
- }
+ @Override
+ public void dropLibrary(MetadataTransactionContext ctx,
+ String dataverseName, String libraryName) throws MetadataException {
+ try {
+ metadataNode
+ .dropLibrary(ctx.getJobId(), dataverseName, libraryName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.dropLibrary(dataverseName, libraryName);
+ }
- @Override
- public void releaseWriteLatch() {
- metadataLatch.writeLock().unlock();
- }
+ @Override
+ public List<Library> getDataverseLibraries(MetadataTransactionContext ctx,
+ String dataverseName) throws MetadataException {
+ List<Library> dataverseLibaries = null;
+ try {
+ // Assuming that the transaction can read its own writes on the
+ // metadata node.
+ dataverseLibaries = metadataNode.getDataverseLibraries(
+ ctx.getJobId(), dataverseName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ // Don't update the cache to avoid checking against the transaction's
+ // uncommitted functions.
+ return dataverseLibaries;
+ }
- @Override
- public void acquireReadLatch() {
- metadataLatch.readLock().lock();
- }
+ @Override
+ public void addLibrary(MetadataTransactionContext ctx, Library library)
+ throws MetadataException {
+ try {
+ metadataNode.addLibrary(ctx.getJobId(), library);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ ctx.addLibrary(library);
+ }
- @Override
- public void releaseReadLatch() {
- metadataLatch.readLock().unlock();
- }
+ @Override
+ public Library getLibrary(MetadataTransactionContext ctx,
+ String dataverseName, String libraryName) throws MetadataException,
+ RemoteException {
+ Library library = null;
+ try {
+ library = metadataNode.getLibrary(ctx.getJobId(), dataverseName,
+ libraryName);
+ } catch (RemoteException e) {
+ throw new MetadataException(e);
+ }
+ return library;
+ }
+
+ @Override
+ public void acquireWriteLatch() {
+ metadataLatch.writeLock().lock();
+ }
+
+ @Override
+ public void releaseWriteLatch() {
+ metadataLatch.writeLock().unlock();
+ }
+
+ @Override
+ public void acquireReadLatch() {
+ metadataLatch.readLock().lock();
+ }
+
+ @Override
+ public void releaseReadLatch() {
+ metadataLatch.readLock().unlock();
+ }
}
\ No newline at end of file
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 d9c6ed4..90b72a2 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
@@ -37,6 +37,7 @@
import edu.uci.ics.asterix.metadata.entities.Function;
import edu.uci.ics.asterix.metadata.entities.Index;
import edu.uci.ics.asterix.metadata.entities.InternalDatasetDetails;
+import edu.uci.ics.asterix.metadata.entities.Library;
import edu.uci.ics.asterix.metadata.entities.Node;
import edu.uci.ics.asterix.metadata.entities.NodeGroup;
import edu.uci.ics.asterix.metadata.entitytupletranslators.DatasetTupleTranslator;
@@ -45,6 +46,7 @@
import edu.uci.ics.asterix.metadata.entitytupletranslators.DataverseTupleTranslator;
import edu.uci.ics.asterix.metadata.entitytupletranslators.FunctionTupleTranslator;
import edu.uci.ics.asterix.metadata.entitytupletranslators.IndexTupleTranslator;
+import edu.uci.ics.asterix.metadata.entitytupletranslators.LibraryTupleTranslator;
import edu.uci.ics.asterix.metadata.entitytupletranslators.NodeGroupTupleTranslator;
import edu.uci.ics.asterix.metadata.entitytupletranslators.NodeTupleTranslator;
import edu.uci.ics.asterix.metadata.valueextractors.DatasetNameValueExtractor;
@@ -265,7 +267,7 @@
ILSMIndex lsmIndex = (ILSMIndex) indexLifecycleManager.getIndex(resourceID);
indexLifecycleManager.open(resourceID);
- //prepare a Callback for logging
+ // prepare a Callback for logging
IModificationOperationCallback modCallback = createIndexModificationCallback(jobId, resourceID, metadataIndex,
lsmIndex, IndexOperation.INSERT);
@@ -376,7 +378,8 @@
searchKey);
deleteTupleFromIndex(jobId, MetadataPrimaryIndexes.DATASET_DATASET, datasetTuple);
} catch (TreeIndexException tie) {
- //ignore this exception and continue deleting all relevant artifacts.
+ // ignore this exception and continue deleting all relevant
+ // artifacts.
}
// Delete entry from secondary index 'group'.
@@ -390,7 +393,8 @@
MetadataSecondaryIndexes.GROUPNAME_ON_DATASET_INDEX, groupNameSearchKey);
deleteTupleFromIndex(jobId, MetadataSecondaryIndexes.GROUPNAME_ON_DATASET_INDEX, groupNameTuple);
} catch (TreeIndexException tie) {
- //ignore this exception and continue deleting all relevant artifacts.
+ // ignore this exception and continue deleting all relevant
+ // artifacts.
}
}
// Delete entry from secondary index 'type'.
@@ -402,7 +406,8 @@
MetadataSecondaryIndexes.DATATYPENAME_ON_DATASET_INDEX, dataTypeSearchKey);
deleteTupleFromIndex(jobId, MetadataSecondaryIndexes.DATATYPENAME_ON_DATASET_INDEX, dataTypeTuple);
} catch (TreeIndexException tie) {
- //ignore this exception and continue deleting all relevant artifacts.
+ // ignore this exception and continue deleting all relevant
+ // artifacts.
}
// Delete entry(s) from the 'indexes' dataset.
@@ -502,7 +507,8 @@
// Searches the index for the tuple to be deleted. Acquires an S
// lock on the 'datatype' dataset.
ITupleReference tuple = getTupleToBeDeleted(jobId, MetadataPrimaryIndexes.DATATYPE_DATASET, searchKey);
- // This call uses the secondary index on datatype. Get nested types before deleting entry from secondary index.
+ // This call uses the secondary index on datatype. Get nested types
+ // before deleting entry from secondary index.
List<String> nestedTypes = getNestedDatatypeNames(jobId, dataverseName, datatypeName);
deleteTupleFromIndex(jobId, MetadataPrimaryIndexes.DATATYPE_DATASET, tuple);
deleteFromDatatypeSecondaryIndex(jobId, dataverseName, datatypeName);
@@ -568,7 +574,7 @@
long resourceID = metadataIndex.getResourceID();
ILSMIndex lsmIndex = (ILSMIndex) indexLifecycleManager.getIndex(resourceID);
indexLifecycleManager.open(resourceID);
- //prepare a Callback for logging
+ // prepare a Callback for logging
IModificationOperationCallback modCallback = createIndexModificationCallback(jobId, resourceID, metadataIndex,
lsmIndex, IndexOperation.DELETE);
IIndexAccessor indexAccessor = lsmIndex.createAccessor(modCallback, NoOpOperationCallback.INSTANCE);
@@ -631,6 +637,20 @@
}
}
+ public List<Library> getDataverseLibraries(JobId jobId, String dataverseName) throws MetadataException,
+ RemoteException {
+ try {
+ ITupleReference searchKey = createTuple(dataverseName);
+ LibraryTupleTranslator tupleReaderWriter = new LibraryTupleTranslator(false);
+ IValueExtractor<Library> valueExtractor = new MetadataEntityValueExtractor<Library>(tupleReaderWriter);
+ List<Library> results = new ArrayList<Library>();
+ searchIndex(jobId, MetadataPrimaryIndexes.LIBRARY_DATASET, searchKey, valueExtractor, results);
+ return results;
+ } catch (Exception e) {
+ throw new MetadataException(e);
+ }
+ }
+
private List<Datatype> getDataverseDatatypes(JobId jobId, String dataverseName) throws MetadataException,
RemoteException {
try {
@@ -853,7 +873,7 @@
return results.get(0);
}
- //Debugging Method
+ // Debugging Method
public String printMetadata() {
StringBuilder sb = new StringBuilder();
@@ -1121,4 +1141,70 @@
throw new MetadataException(e);
}
}
+
+ @Override
+ public void addLibrary(JobId jobId, Library library) throws MetadataException, RemoteException {
+ try {
+ // Insert into the 'Library' dataset.
+ LibraryTupleTranslator tupleReaderWriter = new LibraryTupleTranslator(true);
+ ITupleReference libraryTuple = tupleReaderWriter.getTupleFromMetadataEntity(library);
+ insertTupleIntoIndex(jobId, MetadataPrimaryIndexes.LIBRARY_DATASET, libraryTuple);
+
+ } catch (BTreeDuplicateKeyException e) {
+ throw new MetadataException("A library with this name " + library.getDataverseName()
+ + " already exists in dataverse '" + library.getDataverseName() + "'.", e);
+ } catch (Exception e) {
+ throw new MetadataException(e);
+ }
+
+ }
+
+ @Override
+ public void dropLibrary(JobId jobId, String dataverseName, String libraryName) throws MetadataException,
+ RemoteException {
+ Library library;
+ try {
+ library = getLibrary(jobId, dataverseName, libraryName);
+ } catch (Exception e) {
+ throw new MetadataException(e);
+ }
+ if (library == null) {
+ throw new MetadataException("Cannot drop library '" + library + "' because it doesn't exist.");
+ }
+ try {
+ // Delete entry from the 'Library' dataset.
+ ITupleReference searchKey = createTuple(dataverseName, libraryName);
+ // Searches the index for the tuple to be deleted. Acquires an S
+ // lock on the 'Adapter' dataset.
+ ITupleReference datasetTuple = getTupleToBeDeleted(jobId, MetadataPrimaryIndexes.LIBRARY_DATASET, searchKey);
+ deleteTupleFromIndex(jobId, MetadataPrimaryIndexes.LIBRARY_DATASET, datasetTuple);
+
+ // TODO: Change this to be a BTree specific exception, e.g.,
+ // BTreeKeyDoesNotExistException.
+ } catch (TreeIndexException e) {
+ throw new MetadataException("Cannot drop library '" + libraryName, e);
+ } catch (Exception e) {
+ throw new MetadataException(e);
+ }
+
+ }
+
+ @Override
+ public Library getLibrary(JobId jobId, String dataverseName, String libraryName) throws MetadataException,
+ RemoteException {
+ try {
+ ITupleReference searchKey = createTuple(dataverseName, libraryName);
+ LibraryTupleTranslator tupleReaderWriter = new LibraryTupleTranslator(false);
+ List<Library> results = new ArrayList<Library>();
+ IValueExtractor<Library> valueExtractor = new MetadataEntityValueExtractor<Library>(tupleReaderWriter);
+ searchIndex(jobId, MetadataPrimaryIndexes.LIBRARY_DATASET, searchKey, valueExtractor, results);
+ if (results.isEmpty()) {
+ return null;
+ }
+ return results.get(0);
+ } catch (Exception e) {
+ throw new MetadataException(e);
+ }
+ }
+
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataTransactionContext.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataTransactionContext.java
index cbd37d6..bb0bb8a 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataTransactionContext.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataTransactionContext.java
@@ -26,6 +26,7 @@
import edu.uci.ics.asterix.metadata.entities.Dataverse;
import edu.uci.ics.asterix.metadata.entities.Function;
import edu.uci.ics.asterix.metadata.entities.Index;
+import edu.uci.ics.asterix.metadata.entities.Library;
import edu.uci.ics.asterix.metadata.entities.NodeGroup;
import edu.uci.ics.asterix.transaction.management.service.transaction.JobId;
@@ -123,6 +124,11 @@
logAndApply(new MetadataLogicalOperation(dataverse, false));
}
+ public void addLibrary(Library library) {
+ droppedCache.dropLibrary(library);
+ logAndApply(new MetadataLogicalOperation(library, true));
+ }
+
public void dropDataDatatype(String dataverseName, String datatypeName) {
Datatype datatype = new Datatype(dataverseName, datatypeName, null, false);
droppedCache.addDatatypeIfNotExists(datatype);
@@ -149,6 +155,12 @@
logAndApply(new MetadataLogicalOperation(adapter, false));
}
+ public void dropLibrary(String dataverseName, String libraryName) {
+ Library library = new Library(dataverseName, libraryName);
+ droppedCache.addLibraryIfNotExists(library);
+ logAndApply(new MetadataLogicalOperation(library, false));
+ }
+
public void logAndApply(MetadataLogicalOperation op) {
opLog.add(op);
doOperation(op);
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataManager.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataManager.java
index fdf2d60..46b1f80 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataManager.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataManager.java
@@ -21,16 +21,16 @@
import edu.uci.ics.asterix.common.functions.FunctionSignature;
import edu.uci.ics.asterix.metadata.MetadataException;
import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
-import edu.uci.ics.asterix.metadata.entities.DatasourceAdapter;
import edu.uci.ics.asterix.metadata.entities.Dataset;
+import edu.uci.ics.asterix.metadata.entities.DatasourceAdapter;
import edu.uci.ics.asterix.metadata.entities.Datatype;
import edu.uci.ics.asterix.metadata.entities.Dataverse;
import edu.uci.ics.asterix.metadata.entities.Function;
import edu.uci.ics.asterix.metadata.entities.Index;
+import edu.uci.ics.asterix.metadata.entities.Library;
import edu.uci.ics.asterix.metadata.entities.Node;
import edu.uci.ics.asterix.metadata.entities.NodeGroup;
import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
-import edu.uci.ics.asterix.transaction.management.service.transaction.JobId;
/**
* A metadata manager provides user access to Asterix metadata (e.g., types,
@@ -128,7 +128,7 @@
* @throws MetadataException
*/
List<Dataverse> getDataverses(MetadataTransactionContext ctx) throws MetadataException;
-
+
/**
* Retrieves a dataverse with given name.
*
@@ -179,7 +179,7 @@
* For example, if the dataset already exists.
*/
public void addDataset(MetadataTransactionContext ctx, Dataset dataset) throws MetadataException;
-
+
/**
* Retrieves a dataset within a given dataverse.
*
@@ -441,9 +441,8 @@
public List<Function> getDataverseFunctions(MetadataTransactionContext ctx, String dataverseName)
throws MetadataException;
-
public void initializeDatasetIdFactory(MetadataTransactionContext ctx) throws MetadataException;
-
+
public void acquireWriteLatch();
public void releaseWriteLatch();
@@ -452,5 +451,62 @@
public void releaseReadLatch();
+ /**
+ * Removes a library , acquiring local locks on behalf of the given
+ * transaction id.
+ *
+ * @param ctx
+ * MetadataTransactionContext of an active metadata transaction.
+ * @param dataverseName
+ * dataverse asociated with the adapter that is to be deleted.
+ * @param libraryName
+ * Name of library to be deleted. MetadataException for example,
+ * if the library does not exists.
+ * @throws RemoteException
+ */
+ public void dropLibrary(MetadataTransactionContext ctx, String dataverseName, String libraryName)
+ throws MetadataException;
+
+ /**
+ * Adds a library, acquiring local locks on behalf of the given
+ * transaction id.
+ *
+ * @param ctx
+ * MetadataTransactionContext of an active metadata transaction.
+ * @param library
+ * Library to be added
+ * @throws MetadataException
+ * for example, if the library is already added.
+ * @throws RemoteException
+ */
+ public void addLibrary(MetadataTransactionContext ctx, Library library) throws MetadataException;
+
+ /**
+ * @param ctx
+ * MetadataTransactionContext of an active metadata transaction.
+ * @param dataverseName
+ * dataverse asociated with the library that is to be retrieved.
+ * @param libraryName
+ * name of the library that is to be retrieved
+ * @return Library
+ * @throws MetadataException
+ * @throws RemoteException
+ */
+ public Library getLibrary(MetadataTransactionContext ctx, String dataverseName, String libraryName)
+ throws MetadataException, RemoteException;
+
+ /**
+ * Retireve libraries installed in a given dataverse.
+ *
+ * @param ctx
+ * MetadataTransactionContext of an active metadata transaction.
+ * @param dataverseName
+ * dataverse asociated with the library that is to be retrieved.
+ * @return Library
+ * @throws MetadataException
+ * @throws RemoteException
+ */
+ public List<Library> getDataverseLibraries(MetadataTransactionContext ctx, String dataverseName)
+ throws MetadataException;
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataNode.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataNode.java
index 836d42f..a27dae2 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataNode.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataNode.java
@@ -22,12 +22,13 @@
import edu.uci.ics.asterix.common.functions.FunctionSignature;
import edu.uci.ics.asterix.metadata.MetadataException;
-import edu.uci.ics.asterix.metadata.entities.DatasourceAdapter;
import edu.uci.ics.asterix.metadata.entities.Dataset;
+import edu.uci.ics.asterix.metadata.entities.DatasourceAdapter;
import edu.uci.ics.asterix.metadata.entities.Datatype;
import edu.uci.ics.asterix.metadata.entities.Dataverse;
import edu.uci.ics.asterix.metadata.entities.Function;
import edu.uci.ics.asterix.metadata.entities.Index;
+import edu.uci.ics.asterix.metadata.entities.Library;
import edu.uci.ics.asterix.metadata.entities.Node;
import edu.uci.ics.asterix.metadata.entities.NodeGroup;
import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
@@ -383,7 +384,6 @@
public void addNode(JobId jobId, Node node) throws MetadataException, RemoteException;
/**
-
* @param jobId
* A globally unique id for an active metadata transaction.
* @param functionSignature
@@ -408,7 +408,8 @@
* group to be deleted.
* @throws RemoteException
*/
- public void dropFunction(JobId jobId, FunctionSignature functionSignature) throws MetadataException, RemoteException;
+ public void dropFunction(JobId jobId, FunctionSignature functionSignature) throws MetadataException,
+ RemoteException;
/**
* @param jobId
@@ -442,8 +443,8 @@
public List<DatasourceAdapter> getDataverseAdapters(JobId jobId, String dataverseName) throws MetadataException,
RemoteException;
- public DatasourceAdapter getAdapter(JobId jobId, String dataverseName, String adapterName) throws MetadataException,
- RemoteException;
+ public DatasourceAdapter getAdapter(JobId jobId, String dataverseName, String adapterName)
+ throws MetadataException, RemoteException;
/**
* Deletes a adapter , acquiring local locks on behalf of the given
@@ -474,4 +475,62 @@
public void initializeDatasetIdFactory(JobId jobId) throws MetadataException, RemoteException;
+ /**
+ * Removes a library , acquiring local locks on behalf of the given
+ * transaction id.
+ *
+ * @param txnId
+ * A globally unique id for an active metadata transaction.
+ * @param dataverseName
+ * dataverse asociated with the adapter that is to be deleted.
+ * @param libraryName
+ * Name of library to be deleted. MetadataException for example,
+ * if the library does not exists.
+ * @throws RemoteException
+ */
+ public void dropLibrary(JobId jobId, String dataverseName, String libraryName) throws MetadataException,
+ RemoteException;
+
+ /**
+ * Adds a library, acquiring local locks on behalf of the given
+ * transaction id.
+ *
+ * @param txnId
+ * A globally unique id for an active metadata transaction.
+ * @param library
+ * Library to be added
+ * @throws MetadataException
+ * for example, if the library is already added.
+ * @throws RemoteException
+ */
+ public void addLibrary(JobId jobId, Library library) throws MetadataException, RemoteException;
+
+ /**
+ * @param txnId
+ * A globally unique id for an active metadata transaction.
+ * @param dataverseName
+ * dataverse asociated with the library that is to be retrieved.
+ * @param libraryName
+ * name of the library that is to be retrieved
+ * @return Library
+ * @throws MetadataException
+ * @throws RemoteException
+ */
+ public Library getLibrary(JobId jobId, String dataverseName, String libraryName) throws MetadataException,
+ RemoteException;
+
+ /**
+ * Retireve libraries installed in a given dataverse.
+ *
+ * @param txnId
+ * A globally unique id for an active metadata transaction.
+ * @param dataverseName
+ * dataverse asociated with the library that is to be retrieved.
+ * @return Library
+ * @throws MetadataException
+ * @throws RemoteException
+ */
+ public List<Library> getDataverseLibraries(JobId jobId, String dataverseName) throws MetadataException,
+ RemoteException;
+
}
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 805a550..0bd2a46 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
@@ -119,14 +119,14 @@
MetadataPrimaryIndexes.DATASET_DATASET, MetadataPrimaryIndexes.DATATYPE_DATASET,
MetadataPrimaryIndexes.INDEX_DATASET, MetadataPrimaryIndexes.NODE_DATASET,
MetadataPrimaryIndexes.NODEGROUP_DATASET, MetadataPrimaryIndexes.FUNCTION_DATASET,
- MetadataPrimaryIndexes.DATASOURCE_ADAPTER_DATASET };
+ MetadataPrimaryIndexes.DATASOURCE_ADAPTER_DATASET, MetadataPrimaryIndexes.LIBRARY_DATASET };
secondaryIndexes = new IMetadataIndex[] { MetadataSecondaryIndexes.GROUPNAME_ON_DATASET_INDEX,
MetadataSecondaryIndexes.DATATYPENAME_ON_DATASET_INDEX,
MetadataSecondaryIndexes.DATATYPENAME_ON_DATATYPE_INDEX };
}
- public static void startUniverse(AsterixProperties asterixProperties, INCApplicationContext ncApplicationContext, boolean isNewUniverse)
- throws Exception {
+ public static void startUniverse(AsterixProperties asterixProperties, INCApplicationContext ncApplicationContext,
+ boolean isNewUniverse) throws Exception {
runtimeContext = (AsterixAppRuntimeContext) ncApplicationContext.getApplicationObject();
// Initialize static metadata objects, such as record types and metadata
@@ -233,7 +233,8 @@
primaryIndexes[i].getNodeGroupName());
MetadataManager.INSTANCE.addDataset(mdTxnCtx, new Dataset(primaryIndexes[i].getDataverseName(),
primaryIndexes[i].getIndexedDatasetName(), primaryIndexes[i].getPayloadRecordType().getTypeName(),
- id, new HashMap<String,String>(), DatasetType.INTERNAL, primaryIndexes[i].getDatasetId().getId(), IMetadataEntity.PENDING_NO_OP));
+ id, new HashMap<String, String>(), DatasetType.INTERNAL, primaryIndexes[i].getDatasetId().getId(),
+ IMetadataEntity.PENDING_NO_OP));
}
}
@@ -331,8 +332,8 @@
LSMBTree lsmBtree = null;
long resourceID = -1;
if (create) {
- lsmBtree = LSMBTreeUtils.createLSMTree(memBufferCache, memFreePageManager, ioManager, file,
- bufferCache, fileMapProvider, typeTraits, comparatorFactories, bloomFilterKeyFields,
+ lsmBtree = LSMBTreeUtils.createLSMTree(memBufferCache, memFreePageManager, ioManager, file, bufferCache,
+ fileMapProvider, typeTraits, comparatorFactories, bloomFilterKeyFields,
runtimeContext.getLSMMergePolicy(), runtimeContext.getLSMBTreeOperationTrackerFactory(),
runtimeContext.getLSMIOScheduler(), AsterixRuntimeComponentsProvider.LSMBTREE_PROVIDER);
lsmBtree.create();
@@ -358,13 +359,11 @@
indexLifecycleManager.register(resourceID, lsmBtree);
}
}
-
+
index.setResourceID(resourceID);
index.setFile(file);
indexLifecycleManager.open(resourceID);
}
-
-
public static String getOutputDir() {
return outputDir;
@@ -381,9 +380,9 @@
String datasetName = null;
String indexName = null;
MetadataTransactionContext mdTxnCtx = null;
-
+
MetadataManager.INSTANCE.acquireWriteLatch();
-
+
try {
mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataConstants.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataConstants.java
index 83c147f..2a57b12 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataConstants.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataConstants.java
@@ -19,10 +19,10 @@
// Name of the dataverse the metadata lives in.
public final static String METADATA_DATAVERSE_NAME = "Metadata";
-
+
// Name of the node group where metadata is stored on.
public final static String METADATA_NODEGROUP_NAME = "MetadataGroup";
-
+
// Name of the default nodegroup where internal/feed datasets will be partitioned
// if an explicit nodegroup is not specified at the time of creation of a dataset
public static final String METADATA_DEFAULT_NODEGROUP_NAME = "DEFAULT_NG_ALL_NODES";
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataPrimaryIndexes.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataPrimaryIndexes.java
index 2fc3e59..88cee5b 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataPrimaryIndexes.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataPrimaryIndexes.java
@@ -42,9 +42,12 @@
public static final int NODEGROUP_DATASET_ID = 6;
public static final int FUNCTION_DATASET_ID = 7;
public static final int DATASOURCE_ADAPTER_DATASET_ID = 8;
+ public static final int LIBRARY_DATASET_ID = 9;
+
public static final int FIRST_AVAILABLE_USER_DATASET_ID = 100;
public static IMetadataIndex DATASOURCE_ADAPTER_DATASET;
+ public static IMetadataIndex LIBRARY_DATASET;
/**
* Create all metadata primary index descriptors. MetadataRecordTypes must
@@ -93,5 +96,10 @@
BuiltinType.ASTRING, BuiltinType.ASTRING }, new String[] { "DataverseName", "Name" }, 0,
MetadataRecordTypes.DATASOURCE_ADAPTER_RECORDTYPE, DATASOURCE_ADAPTER_DATASET_ID, true, new int[] { 0,
1 });
+
+ LIBRARY_DATASET = new MetadataIndex("Library", null, 3,
+ new IAType[] { BuiltinType.ASTRING, BuiltinType.ASTRING }, new String[] { "DataverseName", "Name" }, 0,
+ MetadataRecordTypes.LIBRARY_RECORDTYPE, LIBRARY_DATASET_ID, true, new int[] { 0, 1 });
+
}
}
\ No newline at end of file
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java
index 50681ee..fd52aa3 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java
@@ -47,6 +47,7 @@
public static ARecordType NODEGROUP_RECORDTYPE;
public static ARecordType FUNCTION_RECORDTYPE;
public static ARecordType DATASOURCE_ADAPTER_RECORDTYPE;
+ public static ARecordType LIBRARY_RECORDTYPE;
/**
* Create all metadata record types.
@@ -76,6 +77,7 @@
NODEGROUP_RECORDTYPE = createNodeGroupRecordType();
FUNCTION_RECORDTYPE = createFunctionRecordType();
DATASOURCE_ADAPTER_RECORDTYPE = createDatasourceAdapterRecordType();
+ LIBRARY_RECORDTYPE = createLibraryRecordType();
} catch (AsterixException e) {
throw new MetadataException(e);
}
@@ -357,4 +359,13 @@
return new ARecordType("DatasourceAdapterRecordType", fieldNames, fieldTypes, true);
}
+ public static final int LIBRARY_ARECORD_DATAVERSENAME_FIELD_INDEX = 0;
+ public static final int LIBRARY_ARECORD_NAME_FIELD_INDEX = 1;
+ public static final int LIBRARY_ARECORD_TIMESTAMP_FIELD_INDEX = 2;
+
+ private static ARecordType createLibraryRecordType() throws AsterixException {
+ String[] fieldNames = { "DataverseName", "Name", "Timestamp" };
+ IAType[] fieldTypes = { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING };
+ return new ARecordType("LibraryRecordType", fieldNames, fieldTypes, true);
+ }
}
\ No newline at end of file
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/DatasourceAdapter.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/DatasourceAdapter.java
index 2955a08..c360b9f 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/DatasourceAdapter.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/DatasourceAdapter.java
@@ -20,6 +20,8 @@
public class DatasourceAdapter implements IMetadataEntity {
+ private static final long serialVersionUID = 1L;
+
public enum AdapterType {
INTERNAL,
EXTERNAL
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Library.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Library.java
new file mode 100644
index 0000000..a341ba9
--- /dev/null
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Library.java
@@ -0,0 +1,36 @@
+package edu.uci.ics.asterix.metadata.entities;
+
+import edu.uci.ics.asterix.metadata.MetadataCache;
+import edu.uci.ics.asterix.metadata.api.IMetadataEntity;
+
+public class Library implements IMetadataEntity {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String dataverse;
+ private final String name;
+
+ public Library(String dataverseName, String libraryName) {
+ this.dataverse = dataverseName;
+ this.name = libraryName;
+ }
+
+ public String getDataverseName() {
+ return dataverse;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public Object addToCache(MetadataCache cache) {
+ return cache.addLibraryIfNotExists(this);
+ }
+
+ @Override
+ public Object dropFromCache(MetadataCache cache) {
+ return cache.dropLibrary(this);
+ }
+
+}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/LibraryTupleTranslator.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/LibraryTupleTranslator.java
new file mode 100644
index 0000000..4bd2366
--- /dev/null
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/LibraryTupleTranslator.java
@@ -0,0 +1,120 @@
+/*
+ * 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.metadata.entitytupletranslators;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.util.Calendar;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
+import edu.uci.ics.asterix.metadata.MetadataException;
+import edu.uci.ics.asterix.metadata.bootstrap.MetadataPrimaryIndexes;
+import edu.uci.ics.asterix.metadata.bootstrap.MetadataRecordTypes;
+import edu.uci.ics.asterix.metadata.entities.Library;
+import edu.uci.ics.asterix.om.base.ARecord;
+import edu.uci.ics.asterix.om.base.AString;
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+/**
+ * Translates a Library metadata entity to an ITupleReference and vice versa.
+ */
+public class LibraryTupleTranslator extends AbstractTupleTranslator<Library> {
+ // Field indexes of serialized Library in a tuple.
+ // First key field.
+ public static final int LIBRARY_DATAVERSENAME_TUPLE_FIELD_INDEX = 0;
+ // Second key field.
+ public static final int LIBRARY_NAME_TUPLE_FIELD_INDEX = 1;
+
+ // Payload field containing serialized Library.
+ public static final int LIBRARY_PAYLOAD_TUPLE_FIELD_INDEX = 2;
+
+ @SuppressWarnings("unchecked")
+ private ISerializerDeserializer<ARecord> recordSerDes = AqlSerializerDeserializerProvider.INSTANCE
+ .getSerializerDeserializer(MetadataRecordTypes.LIBRARY_RECORDTYPE);
+
+ public LibraryTupleTranslator(boolean getTuple) {
+ super(getTuple, MetadataPrimaryIndexes.LIBRARY_DATASET.getFieldCount());
+ }
+
+ @Override
+ public Library getMetadataEntytiFromTuple(ITupleReference frameTuple) throws IOException {
+ byte[] serRecord = frameTuple.getFieldData(LIBRARY_PAYLOAD_TUPLE_FIELD_INDEX);
+ int recordStartOffset = frameTuple.getFieldStart(LIBRARY_PAYLOAD_TUPLE_FIELD_INDEX);
+ int recordLength = frameTuple.getFieldLength(LIBRARY_PAYLOAD_TUPLE_FIELD_INDEX);
+ ByteArrayInputStream stream = new ByteArrayInputStream(serRecord, recordStartOffset, recordLength);
+ DataInput in = new DataInputStream(stream);
+ ARecord libraryRecord = (ARecord) recordSerDes.deserialize(in);
+ return createLibraryFromARecord(libraryRecord);
+ }
+
+ private Library createLibraryFromARecord(ARecord libraryRecord) {
+ String dataverseName = ((AString) libraryRecord
+ .getValueByPos(MetadataRecordTypes.LIBRARY_ARECORD_DATAVERSENAME_FIELD_INDEX)).getStringValue();
+ String libraryName = ((AString) libraryRecord
+ .getValueByPos(MetadataRecordTypes.LIBRARY_ARECORD_NAME_FIELD_INDEX)).getStringValue();
+
+ return new Library(dataverseName, libraryName);
+ }
+
+ @Override
+ public ITupleReference getTupleFromMetadataEntity(Library library) throws IOException, MetadataException {
+ // write the key in the first 2 fields of the tuple
+ tupleBuilder.reset();
+ aString.setValue(library.getDataverseName());
+ stringSerde.serialize(aString, tupleBuilder.getDataOutput());
+ tupleBuilder.addFieldEndOffset();
+ aString.setValue(library.getName());
+ stringSerde.serialize(aString, tupleBuilder.getDataOutput());
+ tupleBuilder.addFieldEndOffset();
+
+ // write the pay-load in the third field of the tuple
+
+ recordBuilder.reset(MetadataRecordTypes.LIBRARY_RECORDTYPE);
+
+ // write field 0
+ fieldValue.reset();
+ aString.setValue(library.getDataverseName());
+ stringSerde.serialize(aString, fieldValue.getDataOutput());
+ recordBuilder.addField(MetadataRecordTypes.LIBRARY_ARECORD_DATAVERSENAME_FIELD_INDEX, fieldValue);
+
+ // write field 1
+ fieldValue.reset();
+ aString.setValue(library.getName());
+ stringSerde.serialize(aString, fieldValue.getDataOutput());
+ recordBuilder.addField(MetadataRecordTypes.LIBRARY_ARECORD_NAME_FIELD_INDEX, fieldValue);
+
+ // write field 2
+ fieldValue.reset();
+ aString.setValue(Calendar.getInstance().getTime().toString());
+ stringSerde.serialize(aString, fieldValue.getDataOutput());
+ recordBuilder.addField(MetadataRecordTypes.LIBRARY_ARECORD_TIMESTAMP_FIELD_INDEX, fieldValue);
+
+ // write record
+ try {
+ recordBuilder.write(tupleBuilder.getDataOutput(), true);
+ } catch (AsterixException e) {
+ throw new MetadataException(e);
+ }
+ tupleBuilder.addFieldEndOffset();
+ tuple.reset(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray());
+ return tuple;
+ }
+
+}
\ No newline at end of file
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/functions/ExternalFunctionCompilerUtil.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/functions/ExternalFunctionCompilerUtil.java
new file mode 100644
index 0000000..1ad791b
--- /dev/null
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/functions/ExternalFunctionCompilerUtil.java
@@ -0,0 +1,222 @@
+/*
+ * Copyright 2009-2012 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.metadata.functions;
+
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import edu.uci.ics.asterix.external.library.AsterixExternalScalarFunctionInfo;
+import edu.uci.ics.asterix.metadata.MetadataException;
+import edu.uci.ics.asterix.metadata.MetadataManager;
+import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
+import edu.uci.ics.asterix.metadata.entities.Datatype;
+import edu.uci.ics.asterix.metadata.entities.Function;
+import edu.uci.ics.asterix.om.functions.AsterixFunction;
+import edu.uci.ics.asterix.om.typecomputer.base.IResultTypeComputer;
+import edu.uci.ics.asterix.om.typecomputer.impl.ADoubleTypeComputer;
+import edu.uci.ics.asterix.om.typecomputer.impl.AFloatTypeComputer;
+import edu.uci.ics.asterix.om.typecomputer.impl.AInt32TypeComputer;
+import edu.uci.ics.asterix.om.typecomputer.impl.AStringTypeComputer;
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.AUnorderedListType;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression.FunctionKind;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
+import edu.uci.ics.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import edu.uci.ics.hyracks.algebricks.core.algebra.functions.IFunctionInfo;
+import edu.uci.ics.hyracks.algebricks.core.algebra.metadata.IMetadataProvider;
+
+public class ExternalFunctionCompilerUtil implements Serializable {
+
+ private static Pattern orderedListPattern = Pattern.compile("\\[*\\]");
+ private static Pattern unorderedListPattern = Pattern.compile("[{{*}}]");
+
+
+ public static IFunctionInfo getExternalFunctionInfo(MetadataTransactionContext txnCtx, Function function)
+ throws MetadataException {
+
+ String functionKind = function.getKind();
+ IFunctionInfo finfo = null;
+ if (FunctionKind.SCALAR.toString().equalsIgnoreCase(functionKind)) {
+ finfo = getScalarFunctionInfo(txnCtx, function);
+ } else if (FunctionKind.AGGREGATE.toString().equalsIgnoreCase(functionKind)) {
+ finfo = getAggregateFunctionInfo(txnCtx, function);
+ } else if (FunctionKind.STATEFUL.toString().equalsIgnoreCase(functionKind)) {
+ finfo = getStatefulFunctionInfo(txnCtx, function);
+ } else if (FunctionKind.UNNEST.toString().equalsIgnoreCase(functionKind)) {
+ finfo = getUnnestFunctionInfo(txnCtx, function);
+ }
+ return finfo;
+ }
+
+ private static IFunctionInfo getScalarFunctionInfo(MetadataTransactionContext txnCtx, Function function)
+ throws MetadataException {
+ FunctionIdentifier fid = new FunctionIdentifier(function.getDataverseName(), function.getName(),
+ function.getArity());
+ IResultTypeComputer typeComputer = getResultTypeComputer(txnCtx, function);
+ List<IAType> arguments = new ArrayList<IAType>();
+ IAType returnType = null;
+ List<String> paramTypes = function.getParams();
+ for (String paramType : paramTypes) {
+ arguments.add(getTypeInfo(paramType, txnCtx, function));
+ }
+
+ returnType = getTypeInfo(function.getReturnType(), txnCtx, function);
+
+ AsterixExternalScalarFunctionInfo scalarFunctionInfo = new AsterixExternalScalarFunctionInfo(
+ fid.getNamespace(), new AsterixFunction(fid.getName(), fid.getArity()), returnType,
+ function.getFunctionBody(), function.getLanguage(), arguments, typeComputer);
+ return scalarFunctionInfo;
+ }
+
+ private static IAType getTypeInfo(String paramType, MetadataTransactionContext txnCtx, Function function)
+ throws MetadataException {
+ if (paramType.equalsIgnoreCase(BuiltinType.AINT32.getDisplayName())) {
+ return (BuiltinType.AINT32);
+ } else if (paramType.equalsIgnoreCase(BuiltinType.AFLOAT.getDisplayName())) {
+ return (BuiltinType.AFLOAT);
+ } else if (paramType.equalsIgnoreCase(BuiltinType.ASTRING.getDisplayName())) {
+ return (BuiltinType.ASTRING);
+ } else if (paramType.equalsIgnoreCase(BuiltinType.ADOUBLE.getDisplayName())) {
+ return (BuiltinType.ADOUBLE);
+ } else {
+ IAType collection = getCollectionType(paramType, txnCtx, function);
+ if (collection != null) {
+ return collection;
+ } else {
+ Datatype datatype;
+ datatype = MetadataManager.INSTANCE.getDatatype(txnCtx, function.getDataverseName(), paramType);
+ if (datatype == null) {
+ throw new MetadataException(" Type " + paramType + " not defined");
+ }
+ return (datatype.getDatatype());
+ }
+ }
+ }
+
+ private static IAType getCollectionType(String paramType, MetadataTransactionContext txnCtx, Function function)
+ throws MetadataException {
+
+ Matcher matcher = orderedListPattern.matcher(paramType);
+ if (matcher.find()) {
+ String subType = paramType.substring(paramType.indexOf('[') + 1, paramType.lastIndexOf(']'));
+ return new AOrderedListType(getTypeInfo(subType, txnCtx, function), "AOrderedList");
+ } else {
+ matcher = unorderedListPattern.matcher(paramType);
+ if (matcher.find()) {
+ String subType = paramType.substring(paramType.indexOf("{{") + 2, paramType.lastIndexOf("}}"));
+ return new AUnorderedListType(getTypeInfo(subType, txnCtx, function), "AUnorderedList");
+ }
+ }
+ return null;
+ }
+
+ private static IResultTypeComputer getResultTypeComputer(final MetadataTransactionContext txnCtx,
+ final Function function) throws MetadataException {
+
+ final IAType type = getTypeInfo(function.getReturnType(), txnCtx, function);
+ switch (type.getTypeTag()) {
+ case INT32:
+ return AInt32TypeComputer.INSTANCE;
+ case FLOAT:
+ return AFloatTypeComputer.INSTANCE;
+ case DOUBLE:
+ return ADoubleTypeComputer.INSTANCE;
+ case STRING:
+ return AStringTypeComputer.INSTANCE;
+ case ORDEREDLIST:
+ return new IResultTypeComputer() {
+ @Override
+ public IAType computeType(ILogicalExpression expression, IVariableTypeEnvironment env,
+ IMetadataProvider<?, ?> metadataProvider) throws AlgebricksException {
+
+ return new AOrderedListType(((AOrderedListType) type).getItemType(), ((AOrderedListType) type)
+ .getItemType().getTypeName());
+ }
+
+ };
+ case UNORDEREDLIST:
+ return new IResultTypeComputer() {
+ @Override
+ public IAType computeType(ILogicalExpression expression, IVariableTypeEnvironment env,
+ IMetadataProvider<?, ?> metadataProvider) throws AlgebricksException {
+
+ return new AUnorderedListType(type, type.getTypeName());
+ }
+
+ };
+ default:
+ IResultTypeComputer typeComputer = new IResultTypeComputer() {
+ @Override
+ public IAType computeType(ILogicalExpression expression, IVariableTypeEnvironment env,
+ IMetadataProvider<?, ?> mp) throws AlgebricksException {
+ return type;
+ }
+ };
+ return typeComputer;
+ }
+
+ }
+
+ private static IAType getType(Function function, MetadataTransactionContext txnCtx) throws AlgebricksException {
+ IAType collectionType = null;
+ try {
+ collectionType = getCollectionType(function.getReturnType(), txnCtx, function);
+ if (collectionType != null) {
+ return collectionType;
+ } else {
+
+ Datatype datatype;
+ datatype = MetadataManager.INSTANCE.getDatatype(txnCtx, function.getDataverseName(),
+ function.getReturnType());
+ return datatype.getDatatype();
+ }
+ } catch (MetadataException me) {
+ throw new AlgebricksException(me);
+ }
+ }
+
+ private static IFunctionInfo getUnnestFunctionInfo(MetadataTransactionContext txnCtx, Function function) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ private static IFunctionInfo getStatefulFunctionInfo(MetadataTransactionContext txnCtx, Function function) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ private static IFunctionInfo getAggregateFunctionInfo(MetadataTransactionContext txnCtx, Function function) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ public static void main(String args[]) throws FileNotFoundException, IOException {
+ ExternalFunctionCompilerUtil obj = new ExternalFunctionCompilerUtil();
+ ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream("/tmp/ecu.obj"));
+ oos.writeObject(obj);
+ }
+
+}
\ No newline at end of file
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/common/AqlExpressionTypeComputer.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/common/AqlExpressionTypeComputer.java
index 7448a9d..8ddbd99 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/common/AqlExpressionTypeComputer.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/common/AqlExpressionTypeComputer.java
@@ -5,6 +5,7 @@
import edu.uci.ics.asterix.om.constants.AsterixConstantValue;
import edu.uci.ics.asterix.om.functions.AsterixBuiltinFunctions;
+import edu.uci.ics.asterix.om.functions.AsterixExternalFunctionInfo;
import edu.uci.ics.asterix.om.typecomputer.base.IResultTypeComputer;
import edu.uci.ics.asterix.om.types.AUnionType;
import edu.uci.ics.asterix.om.types.BuiltinType;
@@ -58,8 +59,13 @@
unionList.add(BuiltinType.ABOOLEAN);
return new AUnionType(unionList, "OptionalBoolean");
}
- // Note: only builtin functions, for now.
- IResultTypeComputer rtc = AsterixBuiltinFunctions.getResultTypeComputer(fi);
+ // Note: built-in functions + udfs
+ IResultTypeComputer rtc = null;
+ if (AsterixBuiltinFunctions.isBuiltinCompilerFunction(fi)) {
+ rtc = AsterixBuiltinFunctions.getResultTypeComputer(fi);
+ } else {
+ rtc = ((AsterixExternalFunctionInfo) expr.getFunctionInfo()).getResultTypeComputer();
+ }
if (rtc == null) {
throw new AlgebricksException("Type computer missing for " + fi);
}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/functions/AsterixFunctionInfo.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/functions/AsterixFunctionInfo.java
index 4655c37..06d2c39 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/functions/AsterixFunctionInfo.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/functions/AsterixFunctionInfo.java
@@ -14,11 +14,13 @@
*/
package edu.uci.ics.asterix.om.functions;
+import java.io.Serializable;
+
import edu.uci.ics.asterix.common.functions.FunctionSignature;
import edu.uci.ics.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
import edu.uci.ics.hyracks.algebricks.core.algebra.functions.IFunctionInfo;
-public class AsterixFunctionInfo implements IFunctionInfo {
+public class AsterixFunctionInfo implements IFunctionInfo, Serializable {
private final FunctionIdentifier functionIdentifier;
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/base/IResultTypeComputer.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/base/IResultTypeComputer.java
index e90da6b..d3f75d6 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/base/IResultTypeComputer.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/base/IResultTypeComputer.java
@@ -1,5 +1,6 @@
package edu.uci.ics.asterix.om.typecomputer.base;
+import java.io.Serializable;
import edu.uci.ics.asterix.om.types.IAType;
import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -7,7 +8,7 @@
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
import edu.uci.ics.hyracks.algebricks.core.algebra.metadata.IMetadataProvider;
-public interface IResultTypeComputer {
+public interface IResultTypeComputer extends Serializable {
public IAType computeType(ILogicalExpression expression, IVariableTypeEnvironment env,
IMetadataProvider<?, ?> metadataProvider) throws AlgebricksException;
}
