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-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>