You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by dl...@apache.org on 2018/07/20 19:56:31 UTC
[2/4] asterixdb git commit: [NO ISSUE][FUN] Implement array functions
p3
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayAppendDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayAppendDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayAppendDescriptor.java
index aae4d02..6d636ed 100755
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayAppendDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayAppendDescriptor.java
@@ -37,6 +37,20 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.data.std.api.IPointable;
import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+/**
+ * <pre>
+ * array_append(list, val1, val2, ...) returns a new open list with all the values appended to the input list items.
+ * Values can be null (i.e., one can append nulls)
+ *
+ * It throws an error at compile time if the number of arguments < 2
+ *
+ * It returns in order:
+ * 1. missing, if any argument is missing.
+ * 2. null, if the list arg is null or it's not a list.
+ * 3. otherwise, a new open list.
+ *
+ * </pre>
+ */
public class ArrayAppendDescriptor extends AbstractScalarFunctionDynamicDescriptor {
private static final long serialVersionUID = 1L;
private IAType[] argTypes;
@@ -66,7 +80,7 @@ public class ArrayAppendDescriptor extends AbstractScalarFunctionDynamicDescript
@Override
public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
- return new ArrayAppendFunction(args, ctx);
+ return new ArrayAppendEval(args, ctx);
}
};
}
@@ -76,10 +90,9 @@ public class ArrayAppendDescriptor extends AbstractScalarFunctionDynamicDescript
argTypes = (IAType[]) states;
}
- public class ArrayAppendFunction extends AbstractArrayAddRemoveEval {
+ public class ArrayAppendEval extends AbstractArrayAddRemoveEval {
- public ArrayAppendFunction(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx)
- throws HyracksDataException {
+ public ArrayAppendEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
super(args, ctx, 0, 1, args.length - 1, argTypes, false, sourceLoc, true, true);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayConcatDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayConcatDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayConcatDescriptor.java
new file mode 100755
index 0000000..9af159c
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayConcatDescriptor.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.runtime.evaluators.functions;
+
+import org.apache.asterix.builders.IAsterixListBuilder;
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptor;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.functions.IFunctionTypeInferer;
+import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
+import org.apache.asterix.runtime.functions.FunctionTypeInferers;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IPointable;
+
+/**
+ * <pre>
+ * array_concat(list1, list2, ...) returns a new open list with all the values of all lists appended in order into the
+ * new list. Items of the lists can be null or missing (both are added as a null value).
+ *
+ * It throws an error at compile time if the number of arguments < 2
+ *
+ * It returns (or throws an error at runtime) in order:
+ * 1. missing, if any argument is missing.
+ * 2. an error if the input lists are not of the same type (one is an ordered list while the other is unordered).
+ * 3. null, if any input list is null or is not a list.
+ * 4. otherwise, a new open list.
+ *
+ * </pre>
+ */
+public class ArrayConcatDescriptor extends AbstractScalarFunctionDynamicDescriptor {
+ private static final long serialVersionUID = 1L;
+ private IAType[] argTypes;
+
+ public static final IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
+ @Override
+ public IFunctionDescriptor createFunctionDescriptor() {
+ return new ArrayConcatDescriptor();
+ }
+
+ @Override
+ public IFunctionTypeInferer createFunctionTypeInferer() {
+ return FunctionTypeInferers.SET_ARGUMENTS_TYPE;
+ }
+ };
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.ARRAY_CONCAT;
+ }
+
+ @Override
+ public IScalarEvaluatorFactory createEvaluatorFactory(final IScalarEvaluatorFactory[] args)
+ throws AlgebricksException {
+ return new IScalarEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
+ return new ArrayConcatEval(args, ctx);
+ }
+ };
+ }
+
+ @Override
+ public void setImmutableStates(Object... states) {
+ argTypes = (IAType[]) states;
+ }
+
+ public class ArrayConcatEval extends AbstractArrayProcessArraysEval {
+ public ArrayConcatEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
+ super(args, ctx, false, sourceLoc, argTypes);
+ }
+
+ @Override
+ protected void init() {
+ // do nothing
+ }
+
+ @Override
+ protected void finish(IAsterixListBuilder listBuilder) {
+ // do nothing
+ }
+
+ @Override
+ protected void release() {
+ // do nothing
+ }
+
+ @Override
+ protected boolean processItem(IPointable item, int listIndex, IAsterixListBuilder listBuilder)
+ throws HyracksDataException {
+ listBuilder.addItem(item);
+ // listBuilder always copies the item, so we're not using the supplied item
+ return false;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayContainsDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayContainsDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayContainsDescriptor.java
index cc651f6..16bb0c9 100755
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayContainsDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayContainsDescriptor.java
@@ -35,6 +35,21 @@ import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.data.std.api.IPointable;
+/**
+ * <pre>
+ * array_contains(list, val) returns true if the the input list contains the value argument.
+ *
+ * It throws an error at compile time if the number of arguments != 2
+ *
+ * It returns (or throws an error at runtime) in order:
+ * 1. missing, if any argument is missing.
+ * 2. null, if any argument is null.
+ * 3. an error if the value is of a list/object type (i.e. derived type) since deep equality is not yet supported.
+ * 4. null, if the input list is not a list.
+ * 5. otherwise, returns true or false.
+ *
+ * </pre>
+ */
public class ArrayContainsDescriptor extends AbstractScalarFunctionDynamicDescriptor {
private static final long serialVersionUID = 1L;
@@ -58,16 +73,15 @@ public class ArrayContainsDescriptor extends AbstractScalarFunctionDynamicDescri
@Override
public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
- return new ArrayContainsFunction(args, ctx);
+ return new ArrayContainsEval(args, ctx);
}
};
}
- public class ArrayContainsFunction extends AbstractArraySearchEval {
+ public class ArrayContainsEval extends AbstractArraySearchEval {
private final ISerializerDeserializer booleanSerde;
- public ArrayContainsFunction(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx)
- throws HyracksDataException {
+ public ArrayContainsEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
super(args, ctx, sourceLoc);
// TODO(ali): should we get the nontagged serde?
booleanSerde = SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ABOOLEAN);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayDistinctDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayDistinctDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayDistinctDescriptor.java
index 52db331..b74ec6a 100755
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayDistinctDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayDistinctDescriptor.java
@@ -28,7 +28,6 @@ import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
import org.apache.asterix.builders.IAsterixListBuilder;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
-import org.apache.asterix.dataflow.data.nontagged.comparators.AObjectAscBinaryComparatorFactory;
import org.apache.asterix.formats.nontagged.BinaryHashFunctionFactoryProvider;
import org.apache.asterix.om.functions.BuiltinFunctions;
import org.apache.asterix.om.functions.IFunctionDescriptor;
@@ -39,12 +38,12 @@ import org.apache.asterix.om.types.IAType;
import org.apache.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
import org.apache.asterix.runtime.evaluators.common.ListAccessor;
import org.apache.asterix.runtime.functions.FunctionTypeInferers;
+import org.apache.asterix.runtime.utils.ArrayFunctionsUtil;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
import org.apache.hyracks.api.dataflow.value.IBinaryHashFunction;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.SourceLocation;
@@ -92,7 +91,6 @@ public class ArrayDistinctDescriptor extends AbstractScalarFunctionDynamicDescri
}
public class ArrayDistinctFunction extends AbstractArrayProcessEval {
- private final IBinaryComparator comp;
private final SourceLocation sourceLoc;
private final IBinaryHashFunction binaryHashFunction;
private final Int2ObjectMap<List<IPointable>> hashes;
@@ -106,7 +104,6 @@ public class ArrayDistinctDescriptor extends AbstractScalarFunctionDynamicDescri
hashes = new Int2ObjectOpenHashMap<>();
item = pointableAllocator.allocateEmpty();
storage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
- comp = AObjectAscBinaryComparatorFactory.INSTANCE.createBinaryComparator();
binaryHashFunction = BinaryHashFunctionFactoryProvider.INSTANCE.getBinaryHashFunctionFactory(null)
.createBinaryHashFunction();
}
@@ -141,12 +138,10 @@ public class ArrayDistinctDescriptor extends AbstractScalarFunctionDynamicDescri
addItem(item, listBuilder, itemInStorage, sameHashes);
hashes.put(hash, sameHashes);
item = pointableAllocator.allocateEmpty();
- } else {
- // check if it happens that two hashes are the same but they are for different items
- if (isNewItem(item, sameHashes)) {
- addItem(item, listBuilder, itemInStorage, sameHashes);
- item = pointableAllocator.allocateEmpty();
- }
+ } else if (ArrayFunctionsUtil.findItem(item, sameHashes) == null) {
+ // new item, it could happen that two hashes are the same but they are for different items
+ addItem(item, listBuilder, itemInStorage, sameHashes);
+ item = pointableAllocator.allocateEmpty();
}
}
}
@@ -166,16 +161,5 @@ public class ArrayDistinctDescriptor extends AbstractScalarFunctionDynamicDescri
storage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
}
}
-
- private boolean isNewItem(IPointable item, List<IPointable> sameHashes) throws HyracksDataException {
- for (int j = 0; j < sameHashes.size(); j++) {
- if (comp.compare(item.getByteArray(), item.getStartOffset(), item.getLength(),
- sameHashes.get(j).getByteArray(), sameHashes.get(j).getStartOffset(),
- sameHashes.get(j).getLength()) == 0) {
- return false;
- }
- }
- return true;
- }
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayFlattenDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayFlattenDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayFlattenDescriptor.java
new file mode 100644
index 0000000..af55eea
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayFlattenDescriptor.java
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.runtime.evaluators.functions;
+
+import static org.apache.asterix.om.types.EnumDeserializer.ATYPETAGDESERIALIZER;
+
+import java.io.IOException;
+
+import org.apache.asterix.builders.AbvsBuilderFactory;
+import org.apache.asterix.builders.IAsterixListBuilder;
+import org.apache.asterix.builders.OrderedListBuilder;
+import org.apache.asterix.builders.UnorderedListBuilder;
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptor;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.functions.IFunctionTypeInferer;
+import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.AbstractCollectionType;
+import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
+import org.apache.asterix.om.util.container.IObjectPool;
+import org.apache.asterix.om.util.container.ListObjectPool;
+import org.apache.asterix.runtime.base.ListAccessorFactory;
+import org.apache.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
+import org.apache.asterix.runtime.evaluators.common.ListAccessor;
+import org.apache.asterix.runtime.functions.FunctionTypeInferers;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.AbstractPointable;
+import org.apache.hyracks.data.std.api.IMutableValueStorage;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.primitive.TaggedValuePointable;
+import org.apache.hyracks.data.std.primitive.VoidPointable;
+import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+/**
+ * <pre>
+ * array_flatten(list, depth) returns a new open list with any nested list (all types) flattened up to the specified
+ * depth. The returned list type is the same as the input list type. Null and missing items are preserved.
+ * If the depth < 0, then it flattens the input list all the way deep.
+ *
+ * array_flatten([2, null, [5,6], 3, missing], 1) will result in [2, null, 5, 6, 3, null]
+ * array_flatten([2, [5,6], 3], 0) will result in [2, [5,6], 3] (0 depth does nothing)
+ *
+ * It throws an error at compile time if the number of arguments != 2
+ *
+ * It returns in order:
+ * 1. missing, if any argument is missing.
+ * 2. null, if:
+ * - any argument is null.
+ * - the input list is not a list.
+ * - the depth arg is not numeric or
+ * - it's a floating-point number with decimals (e.g. 1.2 will produce null, 1.0 is OK).
+ * 3. otherwise, a new open list.
+ *
+ * </pre>
+ */
+public class ArrayFlattenDescriptor extends AbstractScalarFunctionDynamicDescriptor {
+ private static final long serialVersionUID = 1L;
+ private IAType inputListType;
+
+ public static final IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
+ @Override
+ public IFunctionDescriptor createFunctionDescriptor() {
+ return new ArrayFlattenDescriptor();
+ }
+
+ @Override
+ public IFunctionTypeInferer createFunctionTypeInferer() {
+ return FunctionTypeInferers.SET_ARGUMENT_TYPE;
+ }
+ };
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.ARRAY_FLATTEN;
+ }
+
+ @Override
+ public void setImmutableStates(Object... states) {
+ inputListType = (IAType) states[0];
+ }
+
+ @Override
+ public IScalarEvaluatorFactory createEvaluatorFactory(final IScalarEvaluatorFactory[] args)
+ throws AlgebricksException {
+ return new IScalarEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
+ return new ArrayFlattenEval(args, ctx);
+ }
+ };
+ }
+
+ public class ArrayFlattenEval implements IScalarEvaluator {
+ private final IScalarEvaluator listEval;
+ private final IScalarEvaluator depthEval;
+ private final IPointable list;
+ private final AbstractPointable item;
+ private final TaggedValuePointable depthArg;
+ private final IObjectPool<IMutableValueStorage, ATypeTag> storageAllocator;
+ private final IObjectPool<ListAccessor, ATypeTag> listAccessorAllocator;
+ private final CastTypeEvaluator caster;
+ private ArrayBackedValueStorage storage;
+ private IAsterixListBuilder orderedListBuilder;
+ private IAsterixListBuilder unorderedListBuilder;
+
+ public ArrayFlattenEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
+ storageAllocator = new ListObjectPool<>(new AbvsBuilderFactory());
+ listAccessorAllocator = new ListObjectPool<>(new ListAccessorFactory());
+ storage = new ArrayBackedValueStorage();
+ listEval = args[0].createScalarEvaluator(ctx);
+ depthEval = args[1].createScalarEvaluator(ctx);
+ list = new VoidPointable();
+ item = new VoidPointable();
+ caster = new CastTypeEvaluator();
+ depthArg = new TaggedValuePointable();
+ orderedListBuilder = null;
+ unorderedListBuilder = null;
+ }
+
+ @Override
+ public void evaluate(IFrameTupleReference tuple, IPointable result) throws HyracksDataException {
+ // 1st arg: list to flatten
+ listEval.evaluate(tuple, list);
+ // 2nd arg: depthArg
+ depthEval.evaluate(tuple, depthArg);
+
+ ATypeTag listType = ATYPETAGDESERIALIZER.deserialize(list.getByteArray()[list.getStartOffset()]);
+ if (!ATypeHierarchy.isCompatible(ATYPETAGDESERIALIZER.deserialize(depthArg.getTag()), ATypeTag.DOUBLE)
+ || !listType.isListType()) {
+ PointableHelper.setNull(result);
+ return;
+ }
+ String name = getIdentifier().getName();
+ double depth = ATypeHierarchy.getDoubleValue(name, 1, depthArg.getByteArray(), depthArg.getStartOffset());
+ if (Double.isNaN(depth) || Double.isInfinite(depth) || Math.floor(depth) < depth) {
+ PointableHelper.setNull(result);
+ return;
+ }
+
+ caster.reset(DefaultOpenFieldType.getDefaultOpenFieldType(listType), inputListType, listEval);
+ caster.evaluate(tuple, list);
+
+ int depthInt = (int) depth;
+ // create list
+ IAsterixListBuilder listBuilder;
+ if (listType == ATypeTag.ARRAY) {
+ if (orderedListBuilder == null) {
+ orderedListBuilder = new OrderedListBuilder();
+ }
+ listBuilder = orderedListBuilder;
+ } else {
+ if (unorderedListBuilder == null) {
+ unorderedListBuilder = new UnorderedListBuilder();
+ }
+ listBuilder = unorderedListBuilder;
+ }
+
+ ListAccessor mainListAccessor = listAccessorAllocator.allocate(null);
+ listBuilder.reset((AbstractCollectionType) DefaultOpenFieldType.getDefaultOpenFieldType(listType));
+ mainListAccessor.reset(list.getByteArray(), list.getStartOffset());
+ try {
+ process(mainListAccessor, listBuilder, 0, depthInt);
+ storage.reset();
+ listBuilder.write(storage.getDataOutput(), true);
+ result.set(storage);
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ } finally {
+ storageAllocator.reset();
+ listAccessorAllocator.reset();
+ }
+ }
+
+ private void process(ListAccessor listAccessor, IAsterixListBuilder listBuilder, int currentDepth, int depth)
+ throws IOException {
+ boolean itemInStorage;
+ for (int i = 0; i < listAccessor.size(); i++) {
+ itemInStorage = listAccessor.getOrWriteItem(i, item, storage);
+ // if item is not a list or depth is reached, write it
+ if (!ATYPETAGDESERIALIZER.deserialize(item.getByteArray()[item.getStartOffset()]).isListType()
+ || currentDepth == depth) {
+ listBuilder.addItem(item);
+ } else {
+ // recurse on the sublist
+ ListAccessor newListAccessor = listAccessorAllocator.allocate(null);
+ newListAccessor.reset(item.getByteArray(), item.getStartOffset());
+ if (itemInStorage) {
+ // create a new storage since the item is using it
+ storage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
+ storage.reset();
+ }
+ process(newListAccessor, listBuilder, currentDepth + 1, depth);
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayIfNullDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayIfNullDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayIfNullDescriptor.java
new file mode 100755
index 0000000..53853e2
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayIfNullDescriptor.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.runtime.evaluators.functions;
+
+import static org.apache.asterix.om.types.EnumDeserializer.ATYPETAGDESERIALIZER;
+
+import java.io.IOException;
+
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptor;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
+import org.apache.asterix.runtime.evaluators.common.ListAccessor;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.AbstractPointable;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.primitive.VoidPointable;
+import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+/**
+ * <pre>
+ * array_ifnull(list) returns the first item it encounters that is not a null or missing. Otherwise, it returns null.
+ *
+ * It throws an error at compile time if the number of arguments != 1
+ *
+ * It returns in order:
+ * 1. missing if the input list is missing
+ * 2. null if the input list is null or is not a list.
+ * 3. otherwise, the first non-null non-missing item in the list. Otherwise, null.
+ *
+ * </pre>
+ */
+public class ArrayIfNullDescriptor extends AbstractScalarFunctionDynamicDescriptor {
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
+ @Override
+ public IFunctionDescriptor createFunctionDescriptor() {
+ return new ArrayIfNullDescriptor();
+ }
+ };
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.ARRAY_IFNULL;
+ }
+
+ @Override
+ public IScalarEvaluatorFactory createEvaluatorFactory(final IScalarEvaluatorFactory[] args)
+ throws AlgebricksException {
+ return new IScalarEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
+ return new ArrayIfNullEval(args, ctx);
+ }
+ };
+ }
+
+ public class ArrayIfNullEval implements IScalarEvaluator {
+ private final ArrayBackedValueStorage storage;
+ private final IScalarEvaluator listArgEval;
+ private final IPointable listArg;
+ private final ListAccessor listAccessor;
+ private final AbstractPointable item;
+
+ public ArrayIfNullEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
+ storage = new ArrayBackedValueStorage();
+ listArg = new VoidPointable();
+ item = new VoidPointable();
+ listAccessor = new ListAccessor();
+ listArgEval = args[0].createScalarEvaluator(ctx);
+ }
+
+ @Override
+ public void evaluate(IFrameTupleReference tuple, IPointable result) throws HyracksDataException {
+ // get the list argument and make sure it's a list
+ listArgEval.evaluate(tuple, listArg);
+ byte[] listBytes = listArg.getByteArray();
+ int offset = listArg.getStartOffset();
+ ATypeTag listType = ATYPETAGDESERIALIZER.deserialize(listBytes[offset]);
+ if (!listType.isListType()) {
+ PointableHelper.setNull(result);
+ return;
+ }
+
+ listAccessor.reset(listBytes, offset);
+ ATypeTag itemTypeTag = listAccessor.getItemType();
+ try {
+ if (itemTypeTag == ATypeTag.NULL || itemTypeTag == ATypeTag.MISSING) {
+ // list of nulls or list of missings
+ PointableHelper.setNull(result);
+ return;
+ }
+
+ int numItems = listAccessor.size();
+ for (int i = 0; i < numItems; i++) {
+ listAccessor.getOrWriteItem(i, item, storage);
+ itemTypeTag = ATYPETAGDESERIALIZER.deserialize(item.getByteArray()[item.getStartOffset()]);
+ if (itemTypeTag != ATypeTag.NULL && itemTypeTag != ATypeTag.MISSING) {
+ result.set(item);
+ return;
+ }
+ }
+ PointableHelper.setNull(result);
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayInsertDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayInsertDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayInsertDescriptor.java
index 3d3c54c..634f114 100755
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayInsertDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayInsertDescriptor.java
@@ -41,6 +41,25 @@ import org.apache.hyracks.data.std.api.IPointable;
import org.apache.hyracks.data.std.primitive.TaggedValuePointable;
import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+/**
+ * <pre>
+ * array_insert(list, pos, val1, val2, ...) returns a new open list with all values inserted at the specified position.
+ * Values can be null (i.e., one can insert nulls). Position can be negative where the last position = -1. When position
+ * is positive then the first position = 0. Input list can be empty where the only valid position is 0.
+ * For the list [5,6], the valid positions are 0, 1, 2, -1, -2. If position is floating-point, it's casted to integer.
+ * TODO: should decide on what to do for floating-point positions.
+ *
+ * It throws an error at compile time if the number of arguments < 3
+ *
+ * It returns in order:
+ * 1. missing, if any argument is missing.
+ * 2. null, if
+ * - the list arg is null or it's not a list
+ * - the position is not numeric or the position is out of bound.
+ * 3. otherwise, a new open list.
+ *
+ * </pre>
+ */
public class ArrayInsertDescriptor extends AbstractScalarFunctionDynamicDescriptor {
private static final long serialVersionUID = 1L;
private IAType[] argTypes;
@@ -70,7 +89,7 @@ public class ArrayInsertDescriptor extends AbstractScalarFunctionDynamicDescript
@Override
public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
- return new ArrayInsertFunction(args, ctx);
+ return new ArrayInsertEval(args, ctx);
}
};
}
@@ -80,12 +99,11 @@ public class ArrayInsertDescriptor extends AbstractScalarFunctionDynamicDescript
argTypes = (IAType[]) states;
}
- public class ArrayInsertFunction extends AbstractArrayAddRemoveEval {
+ public class ArrayInsertEval extends AbstractArrayAddRemoveEval {
private final TaggedValuePointable positionArg;
private final IScalarEvaluator positionArgEval;
- public ArrayInsertFunction(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx)
- throws HyracksDataException {
+ public ArrayInsertEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
super(args, ctx, 0, 2, args.length - 2, argTypes, false, sourceLoc, true, true);
positionArg = new TaggedValuePointable();
positionArgEval = args[1].createScalarEvaluator(ctx);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayIntersectDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayIntersectDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayIntersectDescriptor.java
new file mode 100755
index 0000000..f84dee7
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayIntersectDescriptor.java
@@ -0,0 +1,337 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.runtime.evaluators.functions;
+
+import static org.apache.asterix.om.types.EnumDeserializer.ATYPETAGDESERIALIZER;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+
+import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
+import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
+import org.apache.asterix.builders.AbvsBuilderFactory;
+import org.apache.asterix.builders.ArrayListFactory;
+import org.apache.asterix.builders.IAsterixListBuilder;
+import org.apache.asterix.builders.OrderedListBuilder;
+import org.apache.asterix.builders.UnorderedListBuilder;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.common.exceptions.RuntimeDataException;
+import org.apache.asterix.dataflow.data.nontagged.serde.AOrderedListSerializerDeserializer;
+import org.apache.asterix.dataflow.data.nontagged.serde.AUnorderedListSerializerDeserializer;
+import org.apache.asterix.formats.nontagged.BinaryHashFunctionFactoryProvider;
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptor;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.PointableAllocator;
+import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.AbstractCollectionType;
+import org.apache.asterix.om.util.container.IObjectPool;
+import org.apache.asterix.om.util.container.ListObjectPool;
+import org.apache.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
+import org.apache.asterix.runtime.evaluators.common.ListAccessor;
+import org.apache.asterix.runtime.utils.ArrayFunctionsUtil;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.IBinaryHashFunction;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IMutableValueStorage;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.hyracks.data.std.primitive.VoidPointable;
+import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+public class ArrayIntersectDescriptor extends AbstractScalarFunctionDynamicDescriptor {
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
+ @Override
+ public IFunctionDescriptor createFunctionDescriptor() {
+ return new ArrayIntersectDescriptor();
+ }
+ };
+
+ public class ValueListIndex implements IValueReference {
+ private final IPointable value;
+ private int listIndex;
+
+ public ValueListIndex(IPointable value, int listIndex) {
+ this.value = value;
+ this.listIndex = listIndex;
+ }
+
+ @Override
+ public byte[] getByteArray() {
+ return value.getByteArray();
+ }
+
+ @Override
+ public int getStartOffset() {
+ return value.getStartOffset();
+ }
+
+ @Override
+ public int getLength() {
+ return value.getLength();
+ }
+ }
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.ARRAY_INTERSECT;
+ }
+
+ @Override
+ public IScalarEvaluatorFactory createEvaluatorFactory(final IScalarEvaluatorFactory[] args)
+ throws AlgebricksException {
+ return new IScalarEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
+ return new ArrayIntersectEval(args, ctx);
+ }
+ };
+ }
+
+ public class ArrayIntersectEval implements IScalarEvaluator {
+ private final ListAccessor listAccessor;
+ private final IPointable[] listsArgs;
+ private final IScalarEvaluator[] listsEval;
+ private final IBinaryHashFunction binaryHashFunction;
+ private final Int2ObjectMap<List<ValueListIndex>> hashes;
+ private final PointableAllocator pointableAllocator;
+ private final IObjectPool<IMutableValueStorage, ATypeTag> storageAllocator;
+ private final IObjectPool<List<ValueListIndex>, ATypeTag> arrayListAllocator;
+ private final ArrayBackedValueStorage finalResult;
+ private IAsterixListBuilder orderedListBuilder;
+ private IAsterixListBuilder unorderedListBuilder;
+
+ public ArrayIntersectEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
+ orderedListBuilder = null;
+ unorderedListBuilder = null;
+ pointableAllocator = new PointableAllocator();
+ storageAllocator = new ListObjectPool<>(new AbvsBuilderFactory());
+ arrayListAllocator = new ListObjectPool<>(new ArrayListFactory<>());
+ hashes = new Int2ObjectOpenHashMap<>();
+ finalResult = new ArrayBackedValueStorage();
+ listAccessor = new ListAccessor();
+ listsArgs = new IPointable[args.length];
+ listsEval = new IScalarEvaluator[args.length];
+ for (int i = 0; i < args.length; i++) {
+ listsArgs[i] = new VoidPointable();
+ listsEval[i] = args[i].createScalarEvaluator(ctx);
+ }
+ binaryHashFunction = BinaryHashFunctionFactoryProvider.INSTANCE.getBinaryHashFunctionFactory(null)
+ .createBinaryHashFunction();
+ }
+
+ @Override
+ public void evaluate(IFrameTupleReference tuple, IPointable result) throws HyracksDataException {
+ byte listArgType;
+ boolean returnNull = false;
+ AbstractCollectionType outList = null;
+ ATypeTag listTag;
+ int minListIndex = 0;
+ int minSize = -1;
+ int nextSize;
+ IScalarEvaluator listEval;
+ IPointable listArg;
+ // evaluate all the lists first to make sure they're all actually lists and of the same list type
+ for (int i = 0; i < listsEval.length; i++) {
+ listEval = listsEval[i];
+ listEval.evaluate(tuple, listsArgs[i]);
+ if (!returnNull) {
+ listArg = listsArgs[i];
+ listArgType = listArg.getByteArray()[listArg.getStartOffset()];
+ listTag = ATYPETAGDESERIALIZER.deserialize(listArgType);
+ if (!listTag.isListType()) {
+ returnNull = true;
+ } else if (outList != null && outList.getTypeTag() != listTag) {
+ throw new RuntimeDataException(ErrorCode.DIFFERENT_LIST_TYPE_ARGS, sourceLoc);
+ } else {
+ if (outList == null) {
+ outList = (AbstractCollectionType) DefaultOpenFieldType.getDefaultOpenFieldType(listTag);
+ }
+
+ nextSize = getNumItems(outList, listArg.getByteArray(), listArg.getStartOffset());
+ if (nextSize < minSize) {
+ minSize = nextSize;
+ minListIndex = i;
+ }
+ }
+ }
+ }
+
+ if (returnNull) {
+ PointableHelper.setNull(result);
+ return;
+ }
+
+ IAsterixListBuilder listBuilder;
+ if (outList.getTypeTag() == ATypeTag.ARRAY) {
+ if (orderedListBuilder == null) {
+ orderedListBuilder = new OrderedListBuilder();
+ }
+ listBuilder = orderedListBuilder;
+ } else {
+ if (unorderedListBuilder == null) {
+ unorderedListBuilder = new UnorderedListBuilder();
+ }
+ listBuilder = unorderedListBuilder;
+ }
+
+ hashes.clear();
+ try {
+ // first, get distinct items of the most restrictive (smallest) list, pass listBuilder as null since
+ // we're not adding values yet. Values will be added to listBuilder after inspecting all input lists
+ listArg = listsArgs[minListIndex];
+ listAccessor.reset(listArg.getByteArray(), listArg.getStartOffset());
+ processList(listAccessor, minListIndex, null, true);
+
+ // now process each list one by one
+ listBuilder.reset(outList);
+ for (int listIndex = 0; listIndex < listsArgs.length; listIndex++) {
+ if (listIndex == minListIndex) {
+ incrementSmallest(listIndex, hashes.values());
+ } else {
+ listArg = listsArgs[listIndex];
+ listAccessor.reset(listArg.getByteArray(), listArg.getStartOffset());
+ processList(listAccessor, listIndex, listBuilder, false);
+ }
+ }
+
+ finalResult.reset();
+ listBuilder.write(finalResult.getDataOutput(), true);
+ result.set(finalResult);
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ } finally {
+ storageAllocator.reset();
+ arrayListAllocator.reset();
+ pointableAllocator.reset();
+ }
+ }
+
+ private int getNumItems(AbstractCollectionType listType, byte[] listBytes, int offset) {
+ if (listType.getTypeTag() == ATypeTag.ARRAY) {
+ return AOrderedListSerializerDeserializer.getNumberOfItems(listBytes, offset);
+ } else {
+ return AUnorderedListSerializerDeserializer.getNumberOfItems(listBytes, offset);
+ }
+ }
+
+ private void processList(ListAccessor listAccessor, int listIndex, IAsterixListBuilder listBuilder,
+ boolean initIntersectList) throws IOException {
+ int hash;
+ List<ValueListIndex> sameHashes;
+ boolean itemInStorage;
+ IPointable item = pointableAllocator.allocateEmpty();
+ ArrayBackedValueStorage storage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
+ storage.reset();
+ for (int j = 0; j < listAccessor.size(); j++) {
+ itemInStorage = listAccessor.getOrWriteItem(j, item, storage);
+ if (ATYPETAGDESERIALIZER.deserialize(item.getByteArray()[item.getStartOffset()]).isDerivedType()) {
+ throw new RuntimeDataException(ErrorCode.CANNOT_COMPARE_COMPLEX, sourceLoc);
+ }
+ if (notNullAndMissing(item)) {
+ // look up to see if item exists
+ hash = binaryHashFunction.hash(item.getByteArray(), item.getStartOffset(), item.getLength());
+ sameHashes = hashes.get(hash);
+ if (initIntersectList && initIntersectList(item, hash, sameHashes)) {
+ // item is used
+ item = pointableAllocator.allocateEmpty();
+ if (itemInStorage) {
+ storage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
+ storage.reset();
+ }
+ } else {
+ incrementCommonValue(item, sameHashes, listIndex, listBuilder);
+ }
+ }
+ }
+ }
+
+ // collect the items of the most restrictive list, it initializes the list index as -1. each successive list
+ // should stamp the value with its list index if the list has the item. It starts with list index = 0
+ private boolean initIntersectList(IPointable item, int hash, List<ValueListIndex> sameHashes)
+ throws IOException {
+ // add if new item
+ if (sameHashes == null) {
+ List<ValueListIndex> newHashes = arrayListAllocator.allocate(null);
+ newHashes.clear();
+ newHashes.add(new ValueListIndex(item, -1));
+ hashes.put(hash, newHashes);
+ return true;
+ } else if (ArrayFunctionsUtil.findItem(item, sameHashes) == null) {
+ sameHashes.add(new ValueListIndex(item, -1));
+ return true;
+ }
+ // else ignore for duplicate values in the same list
+ return false;
+ }
+
+ private void incrementCommonValue(IPointable item, List<ValueListIndex> sameHashes, int listIndex,
+ IAsterixListBuilder listBuilder) throws IOException {
+ if (sameHashes != null) {
+ // look for the same equal item, add to list builder when all lists have seen this item
+ incrementIfExists(sameHashes, item, listIndex, listBuilder);
+ }
+ }
+
+ private boolean notNullAndMissing(IPointable item) {
+ byte tag = item.getByteArray()[item.getStartOffset()];
+ return tag != ATypeTag.SERIALIZED_NULL_TYPE_TAG && tag != ATypeTag.SERIALIZED_MISSING_TYPE_TAG;
+ }
+
+ // this method is only for the most restrictive list. it avoids comparison since it is the initial list we start
+ // with, so for sure every element in the collection must exist in the list
+ private void incrementSmallest(int listIndex, Collection<List<ValueListIndex>> commonValues) {
+ for (List<ValueListIndex> items : commonValues) {
+ for (int i = 0; i < items.size(); i++) {
+ // any difference that is not == 1 means either this current list has already stamped and advanced
+ // the stamp or the item is not common among lists because if it's common then each list should've
+ // incremented the item list index up to the current list index
+ if (listIndex - items.get(i).listIndex == 1) {
+ items.get(i).listIndex = listIndex;
+ }
+ }
+ }
+ }
+
+ private void incrementIfExists(List<ValueListIndex> sameHashes, IPointable item, int listIndex,
+ IAsterixListBuilder listBuilder) throws HyracksDataException {
+ ValueListIndex sameValue = ArrayFunctionsUtil.findItem(item, sameHashes);
+ if (sameValue != null && listIndex - sameValue.listIndex == 1) {
+ // found the item, its stamp is OK (stamp saves the last list index that has seen this item)
+ // increment stamp of this item
+ sameValue.listIndex = listIndex;
+ if (listIndex == listsArgs.length - 1) {
+ // when listIndex is the last list, then it means this item was found in all previous lists
+ listBuilder.addItem(item);
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPositionDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPositionDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPositionDescriptor.java
index 5258abd..e4e54f1 100755
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPositionDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPositionDescriptor.java
@@ -34,6 +34,22 @@ import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.data.std.api.IPointable;
+/**
+ * <pre>
+ * array_position(list, val) returns the 0-based position (as integer) of the value argument in the input list. If the
+ * value does not exists, it returns -1
+ *
+ * It throws an error at compile time if the number of arguments != 2
+ *
+ * It returns (or throws an error at runtime) in order:
+ * 1. missing, if any argument is missing.
+ * 2. null, if any argument is null.
+ * 3. an error if the value is of a list/object type (i.e. derived type) since deep equality is not yet supported.
+ * 4. null, if the input list is not a list.
+ * 5. otherwise, returns the position of the value in the list or -1 if not found.
+ *
+ * </pre>
+ */
public class ArrayPositionDescriptor extends AbstractScalarFunctionDynamicDescriptor {
private static final long serialVersionUID = 1L;
@@ -57,16 +73,15 @@ public class ArrayPositionDescriptor extends AbstractScalarFunctionDynamicDescri
@Override
public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
- return new ArrayPositionFunction(args, ctx);
+ return new ArrayPositionEval(args, ctx);
}
};
}
- public class ArrayPositionFunction extends AbstractArraySearchEval {
+ public class ArrayPositionEval extends AbstractArraySearchEval {
private final ISerializerDeserializer intSerde;
- public ArrayPositionFunction(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx)
- throws HyracksDataException {
+ public ArrayPositionEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
super(args, ctx, sourceLoc);
intSerde = SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AINT32);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPrependDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPrependDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPrependDescriptor.java
index 7e52df5..dcb6ad8 100755
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPrependDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPrependDescriptor.java
@@ -35,6 +35,20 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.data.std.api.IPointable;
import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+/**
+ * <pre>
+ * array_prepend(val1, val2, ..., list) returns a new open list with all the values prepended to the input list items.
+ * Values can be null (i.e., one can append nulls)
+ *
+ * It throws an error at compile time if the number of arguments < 2
+ *
+ * It returns in order:
+ * 1. missing, if any argument is missing.
+ * 2. null, if the list arg is null or it's not a list.
+ * 3. otherwise, a new open list.
+ *
+ * </pre>
+ */
public class ArrayPrependDescriptor extends AbstractScalarFunctionDynamicDescriptor {
private static final long serialVersionUID = 1L;
private IAType[] argTypes;
@@ -64,7 +78,7 @@ public class ArrayPrependDescriptor extends AbstractScalarFunctionDynamicDescrip
@Override
public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
- return new ArrayPrependFunction(args, ctx);
+ return new ArrayPrependEval(args, ctx);
}
};
}
@@ -74,10 +88,9 @@ public class ArrayPrependDescriptor extends AbstractScalarFunctionDynamicDescrip
argTypes = (IAType[]) states;
}
- public class ArrayPrependFunction extends AbstractArrayAddRemoveEval {
+ public class ArrayPrependEval extends AbstractArrayAddRemoveEval {
- public ArrayPrependFunction(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx)
- throws HyracksDataException {
+ public ArrayPrependEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
super(args, ctx, args.length - 1, 0, args.length - 1, argTypes, false, sourceLoc, true, true);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPutDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPutDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPutDescriptor.java
index 6435369..fc17e99 100755
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPutDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayPutDescriptor.java
@@ -44,6 +44,22 @@ import org.apache.hyracks.data.std.api.IPointable;
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+/**
+ * <pre>
+ * array_put(list, val1, val2, ...) returns a new open list with all the values appended to the input list items only if
+ * the list does not already have the value. Values cannot be null (i.e., one cannot append nulls).
+ * array_put([2, 3], 2, 2, 9, 9) will result in [2, 3, 9, 9].
+ *
+ * It throws an error at compile time if the number of arguments < 2
+ *
+ * It returns (or throws an error at runtime) in order:
+ * 1. missing, if any argument is missing.
+ * 2. null, if any argument is null.
+ * 3. an error if any value arg is of a list/object type (i.e. derived type) since deep equality is not yet supported.
+ * 4. otherwise, a new open list.
+ *
+ * </pre>
+ */
public class ArrayPutDescriptor extends AbstractScalarFunctionDynamicDescriptor {
private static final long serialVersionUID = 1L;
private IAType[] argTypes;
@@ -73,7 +89,7 @@ public class ArrayPutDescriptor extends AbstractScalarFunctionDynamicDescriptor
@Override
public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
- return new ArrayPutFunction(args, ctx);
+ return new ArrayPutEval(args, ctx);
}
};
}
@@ -83,11 +99,11 @@ public class ArrayPutDescriptor extends AbstractScalarFunctionDynamicDescriptor
argTypes = (IAType[]) states;
}
- public class ArrayPutFunction extends AbstractArrayAddRemoveEval {
+ public class ArrayPutEval extends AbstractArrayAddRemoveEval {
private final ArrayBackedValueStorage storage;
private final IBinaryComparator comp;
- public ArrayPutFunction(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
+ public ArrayPutEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
super(args, ctx, 0, 1, args.length - 1, argTypes, true, sourceLoc, true, false);
comp = AObjectAscBinaryComparatorFactory.INSTANCE.createBinaryComparator();
storage = new ArrayBackedValueStorage();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRangeDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRangeDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRangeDescriptor.java
new file mode 100755
index 0000000..4068101
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRangeDescriptor.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.runtime.evaluators.functions;
+
+import static org.apache.asterix.om.types.EnumDeserializer.ATYPETAGDESERIALIZER;
+
+import org.apache.asterix.builders.OrderedListBuilder;
+import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
+import org.apache.asterix.om.base.AMutableDouble;
+import org.apache.asterix.om.base.AMutableInt64;
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptor;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.typecomputer.impl.ArrayRangeTypeComputer;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.BuiltinType;
+import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
+import org.apache.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.primitive.TaggedValuePointable;
+import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+public class ArrayRangeDescriptor extends AbstractScalarFunctionDynamicDescriptor {
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
+ @Override
+ public IFunctionDescriptor createFunctionDescriptor() {
+ return new ArrayRangeDescriptor();
+ }
+ };
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.ARRAY_RANGE;
+ }
+
+ @Override
+ public IScalarEvaluatorFactory createEvaluatorFactory(final IScalarEvaluatorFactory[] args)
+ throws AlgebricksException {
+ return new IScalarEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
+ return new ArrayRangeEval(args, ctx);
+ }
+ };
+ }
+
+ public class ArrayRangeEval implements IScalarEvaluator {
+ private final OrderedListBuilder listBuilder;
+ private final ArrayBackedValueStorage storage;
+ private final IScalarEvaluator startNumEval;
+ private final TaggedValuePointable start;
+ private final IScalarEvaluator endNumEval;
+ private final TaggedValuePointable end;
+ private final AMutableDouble aDouble;
+ private final AMutableInt64 aLong;
+ private IScalarEvaluator stepNumEval;
+ private TaggedValuePointable step;
+
+ public ArrayRangeEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
+ storage = new ArrayBackedValueStorage();
+ start = new TaggedValuePointable();
+ end = new TaggedValuePointable();
+ startNumEval = args[0].createScalarEvaluator(ctx);
+ endNumEval = args[1].createScalarEvaluator(ctx);
+ listBuilder = new OrderedListBuilder();
+ aDouble = new AMutableDouble(0);
+ aLong = new AMutableInt64(0);
+ if (args.length == 3) {
+ stepNumEval = args[2].createScalarEvaluator(ctx);
+ step = new TaggedValuePointable();
+ }
+ }
+
+ @Override
+ public void evaluate(IFrameTupleReference tuple, IPointable result) throws HyracksDataException {
+ startNumEval.evaluate(tuple, start);
+ endNumEval.evaluate(tuple, end);
+ String n = getIdentifier().getName();
+ ATypeTag startTag = ATYPETAGDESERIALIZER.deserialize(start.getTag());
+ ATypeTag endTag = ATYPETAGDESERIALIZER.deserialize(end.getTag());
+ ATypeTag stepTag = ATypeTag.INTEGER;
+ double stepNum = 1;
+ if (stepNumEval != null) {
+ stepNumEval.evaluate(tuple, step);
+ stepTag = ATYPETAGDESERIALIZER.deserialize(step.getTag());
+ if (!ATypeHierarchy.isCompatible(ATypeTag.DOUBLE, stepTag)) {
+ PointableHelper.setNull(result);
+ return;
+ }
+ stepNum = ATypeHierarchy.getDoubleValue(n, 2, step.getByteArray(), step.getStartOffset());
+ }
+
+ if (!ATypeHierarchy.isCompatible(ATypeTag.DOUBLE, startTag)
+ || !ATypeHierarchy.isCompatible(ATypeTag.DOUBLE, endTag)) {
+ PointableHelper.setNull(result);
+ return;
+ }
+
+ ISerializerDeserializer serde;
+ if (ATypeHierarchy.canPromote(startTag, ATypeTag.BIGINT)
+ && ATypeHierarchy.canPromote(endTag, ATypeTag.BIGINT)
+ && ATypeHierarchy.canPromote(stepTag, ATypeTag.BIGINT)) {
+ // all 3 numbers are whole numbers
+ serde = SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AINT64);
+ long startNum = ATypeHierarchy.getLongValue(n, 0, start.getByteArray(), start.getStartOffset());
+ long endNum = ATypeHierarchy.getLongValue(n, 1, end.getByteArray(), end.getStartOffset());
+ listBuilder.reset(ArrayRangeTypeComputer.LONG_LIST);
+ while ((startNum < endNum && stepNum > 0) || (startNum > endNum && stepNum < 0)) {
+ aLong.setValue(startNum);
+ storage.reset();
+ serde.serialize(aLong, storage.getDataOutput());
+ listBuilder.addItem(storage);
+ startNum += stepNum;
+ }
+ } else {
+ // one number is a floating-point number
+ serde = SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ADOUBLE);
+ double startNum = ATypeHierarchy.getDoubleValue(n, 0, start.getByteArray(), start.getStartOffset());
+ double endNum = ATypeHierarchy.getDoubleValue(n, 1, end.getByteArray(), end.getStartOffset());
+ listBuilder.reset(ArrayRangeTypeComputer.DOUBLE_LIST);
+ while ((startNum < endNum && stepNum > 0) || (startNum > endNum && stepNum < 0)) {
+ aDouble.setValue(startNum);
+ storage.reset();
+ serde.serialize(aDouble, storage.getDataOutput());
+ listBuilder.addItem(storage);
+ startNum += stepNum;
+ }
+ }
+
+ storage.reset();
+ listBuilder.write(storage.getDataOutput(), true);
+ result.set(storage);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRemoveDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRemoveDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRemoveDescriptor.java
index ff7ace1..ea27017 100755
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRemoveDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRemoveDescriptor.java
@@ -42,6 +42,21 @@ import org.apache.hyracks.data.std.api.IPointable;
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+/**
+ * <pre>
+ * array_remove(list, val1, val2, ...) returns a new (open or closed) list with all the values removed from the input
+ * list. Values cannot be null (i.e., one cannot remove nulls).
+ *
+ * It throws an error at compile time if the number of arguments < 2
+ *
+ * It returns (or throws an error at runtime) in order:
+ * 1. missing, if any argument is missing.
+ * 2. null, if any argument is null.
+ * 4. an error if any value arg is of a list/object type (i.e. derived type) since deep equality is not yet supported.
+ * 3. otherwise, a new list that has the same type as the input list.
+ *
+ * </pre>
+ */
public class ArrayRemoveDescriptor extends AbstractScalarFunctionDynamicDescriptor {
private static final long serialVersionUID = 1L;
private IAType[] argTypes;
@@ -71,7 +86,7 @@ public class ArrayRemoveDescriptor extends AbstractScalarFunctionDynamicDescript
@Override
public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
- return new ArrayRemoveFunction(args, ctx);
+ return new ArrayRemoveEval(args, ctx);
}
};
}
@@ -81,12 +96,11 @@ public class ArrayRemoveDescriptor extends AbstractScalarFunctionDynamicDescript
argTypes = (IAType[]) states;
}
- public class ArrayRemoveFunction extends AbstractArrayAddRemoveEval {
+ public class ArrayRemoveEval extends AbstractArrayAddRemoveEval {
private final ArrayBackedValueStorage storage;
private final IBinaryComparator comp;
- public ArrayRemoveFunction(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx)
- throws HyracksDataException {
+ public ArrayRemoveEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
super(args, ctx, 0, 1, args.length - 1, argTypes, true, sourceLoc, false, false);
storage = new ArrayBackedValueStorage();
comp = AObjectAscBinaryComparatorFactory.INSTANCE.createBinaryComparator();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRepeatDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRepeatDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRepeatDescriptor.java
index 3a0280d..ce9d3cc 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRepeatDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayRepeatDescriptor.java
@@ -18,8 +18,6 @@
*/
package org.apache.asterix.runtime.evaluators.functions;
-import static org.apache.asterix.om.types.AOrderedListType.FULL_OPEN_ORDEREDLIST_TYPE;
-
import org.apache.asterix.builders.IAsterixListBuilder;
import org.apache.asterix.builders.OrderedListBuilder;
import org.apache.asterix.om.functions.BuiltinFunctions;
@@ -80,12 +78,12 @@ public class ArrayRepeatDescriptor extends AbstractScalarFunctionDynamicDescript
@Override
public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
- return new ArrayRepeatFunction(args, ctx);
+ return new ArrayRepeatEval(args, ctx);
}
};
}
- public class ArrayRepeatFunction implements IScalarEvaluator {
+ public class ArrayRepeatEval implements IScalarEvaluator {
private final ArrayBackedValueStorage storage;
private final IScalarEvaluator repeatedValueEval;
private final IScalarEvaluator repeatEval;
@@ -94,8 +92,7 @@ public class ArrayRepeatDescriptor extends AbstractScalarFunctionDynamicDescript
private final TaggedValuePointable repeatArgValue;
private final IAsterixListBuilder listBuilder;
- public ArrayRepeatFunction(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx)
- throws HyracksDataException {
+ public ArrayRepeatEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
storage = new ArrayBackedValueStorage();
repeatedValueEval = args[0].createScalarEvaluator(ctx);
repeatEval = args[1].createScalarEvaluator(ctx);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/9a987680/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayReplaceDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayReplaceDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayReplaceDescriptor.java
new file mode 100644
index 0000000..e8d77a8
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/ArrayReplaceDescriptor.java
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.runtime.evaluators.functions;
+
+import static org.apache.asterix.om.types.EnumDeserializer.ATYPETAGDESERIALIZER;
+
+import java.io.IOException;
+
+import org.apache.asterix.builders.IAsterixListBuilder;
+import org.apache.asterix.builders.OrderedListBuilder;
+import org.apache.asterix.builders.UnorderedListBuilder;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.common.exceptions.RuntimeDataException;
+import org.apache.asterix.dataflow.data.nontagged.comparators.AObjectAscBinaryComparatorFactory;
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptor;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.functions.IFunctionTypeInferer;
+import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.AbstractCollectionType;
+import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
+import org.apache.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
+import org.apache.asterix.runtime.evaluators.common.ListAccessor;
+import org.apache.asterix.runtime.functions.FunctionTypeInferers;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.AbstractPointable;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.primitive.TaggedValuePointable;
+import org.apache.hyracks.data.std.primitive.VoidPointable;
+import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+public class ArrayReplaceDescriptor extends AbstractScalarFunctionDynamicDescriptor {
+ private static final long serialVersionUID = 1L;
+ private IAType inputListType;
+ private IAType newValueType;
+
+ public static final IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
+ @Override
+ public IFunctionDescriptor createFunctionDescriptor() {
+ return new ArrayReplaceDescriptor();
+ }
+
+ @Override
+ public IFunctionTypeInferer createFunctionTypeInferer() {
+ return FunctionTypeInferers.SET_ARGUMENTS_TYPE;
+ }
+ };
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.ARRAY_REPLACE;
+ }
+
+ @Override
+ public void setImmutableStates(Object... states) {
+ inputListType = (IAType) states[0];
+ newValueType = (IAType) states[2];
+ }
+
+ @Override
+ public IScalarEvaluatorFactory createEvaluatorFactory(final IScalarEvaluatorFactory[] args)
+ throws AlgebricksException {
+ return new IScalarEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
+ return new ArrayReplaceEval(args, ctx);
+ }
+ };
+ }
+
+ public class ArrayReplaceEval implements IScalarEvaluator {
+ private final IScalarEvaluator listEval;
+ private final IScalarEvaluator targetValEval;
+ private final IScalarEvaluator newValEval;
+ private IScalarEvaluator maxEval;
+ private final IPointable list;
+ private final IPointable target;
+ private final IPointable newVal;
+ private TaggedValuePointable maxArg;
+ private final AbstractPointable item;
+ private final ListAccessor listAccessor;
+ private final IBinaryComparator comp;
+ private final ArrayBackedValueStorage storage;
+ private final CastTypeEvaluator caster;
+ private IAsterixListBuilder orderedListBuilder;
+ private IAsterixListBuilder unorderedListBuilder;
+
+ public ArrayReplaceEval(IScalarEvaluatorFactory[] args, IHyracksTaskContext ctx) throws HyracksDataException {
+ storage = new ArrayBackedValueStorage();
+ listEval = args[0].createScalarEvaluator(ctx);
+ targetValEval = args[1].createScalarEvaluator(ctx);
+ newValEval = args[2].createScalarEvaluator(ctx);
+ if (args.length == 4) {
+ maxEval = args[3].createScalarEvaluator(ctx);
+ maxArg = new TaggedValuePointable();
+ }
+ list = new VoidPointable();
+ target = new VoidPointable();
+ newVal = new VoidPointable();
+ item = new VoidPointable();
+ listAccessor = new ListAccessor();
+ caster = new CastTypeEvaluator();
+ orderedListBuilder = null;
+ unorderedListBuilder = null;
+ comp = AObjectAscBinaryComparatorFactory.INSTANCE.createBinaryComparator();
+ }
+
+ @Override
+ public void evaluate(IFrameTupleReference tuple, IPointable result) throws HyracksDataException {
+ storage.reset();
+ listEval.evaluate(tuple, list);
+ targetValEval.evaluate(tuple, target);
+ newValEval.evaluate(tuple, newVal);
+ ATypeTag listType = ATYPETAGDESERIALIZER.deserialize(list.getByteArray()[list.getStartOffset()]);
+ ATypeTag targetTag = ATYPETAGDESERIALIZER.deserialize(target.getByteArray()[target.getStartOffset()]);
+ ATypeTag newValTag = ATYPETAGDESERIALIZER.deserialize(newVal.getByteArray()[newVal.getStartOffset()]);
+ if (listType == ATypeTag.MISSING || targetTag == ATypeTag.MISSING || newValTag == ATypeTag.MISSING) {
+ PointableHelper.setMissing(result);
+ return;
+ }
+
+ double maxDouble = -1;
+ String name = getIdentifier().getName();
+ if (maxEval != null) {
+ maxEval.evaluate(tuple, maxArg);
+ ATypeTag maxTag = ATYPETAGDESERIALIZER.deserialize(maxArg.getTag());
+ if (maxTag == ATypeTag.MISSING) {
+ PointableHelper.setMissing(result);
+ return;
+ } else if (!ATypeHierarchy.isCompatible(maxTag, ATypeTag.DOUBLE)) {
+ PointableHelper.setNull(result);
+ return;
+ }
+ maxDouble = ATypeHierarchy.getDoubleValue(name, 3, maxArg.getByteArray(), maxArg.getStartOffset());
+ }
+
+ if (!listType.isListType() || Math.floor(maxDouble) < maxDouble || targetTag == ATypeTag.NULL
+ || Double.isInfinite(maxDouble) || Double.isNaN(maxDouble)) {
+ PointableHelper.setNull(result);
+ return;
+ }
+
+ if (targetTag.isDerivedType()) {
+ throw new RuntimeDataException(ErrorCode.CANNOT_COMPARE_COMPLEX, sourceLoc);
+ }
+
+ IAType defaultOpenType = DefaultOpenFieldType.getDefaultOpenFieldType(listType);
+ caster.reset(defaultOpenType, inputListType, listEval);
+ caster.evaluate(tuple, list);
+
+ defaultOpenType = DefaultOpenFieldType.getDefaultOpenFieldType(newValTag);
+ if (defaultOpenType != null) {
+ caster.reset(defaultOpenType, newValueType, newValEval);
+ caster.evaluate(tuple, newVal);
+ }
+
+ int max = (int) maxDouble;
+ // create list
+ IAsterixListBuilder listBuilder;
+ if (listType == ATypeTag.ARRAY) {
+ if (orderedListBuilder == null) {
+ orderedListBuilder = new OrderedListBuilder();
+ }
+ listBuilder = orderedListBuilder;
+ } else {
+ if (unorderedListBuilder == null) {
+ unorderedListBuilder = new UnorderedListBuilder();
+ }
+ listBuilder = unorderedListBuilder;
+ }
+
+ listBuilder.reset((AbstractCollectionType) DefaultOpenFieldType.getDefaultOpenFieldType(listType));
+ listAccessor.reset(list.getByteArray(), list.getStartOffset());
+ try {
+ int counter = 0;
+ byte[] targetBytes = target.getByteArray();
+ int offset = target.getStartOffset();
+ int length = target.getLength();
+ for (int i = 0; i < listAccessor.size(); i++) {
+ listAccessor.getOrWriteItem(i, item, storage);
+ if (counter != max && comp.compare(item.getByteArray(), item.getStartOffset(), item.getLength(),
+ targetBytes, offset, length) == 0) {
+ listBuilder.addItem(newVal);
+ counter++;
+ } else {
+ listBuilder.addItem(item);
+ }
+ }
+ storage.reset();
+ listBuilder.write(storage.getDataOutput(), true);
+ result.set(storage);
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+ }
+}