You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2024/03/26 21:22:48 UTC
Change in asterixdb[neo]: [WIP] median
From Ali Alsuliman <al...@gmail.com>:
Ali Alsuliman has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/18210 )
Change subject: [WIP] median
......................................................................
[WIP] median
Change-Id: Id8d03f42d54b5ed4cf316c31f0b4cce9dd7c1dc0
---
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/MedianAggregateDescriptor.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/MedianAggregateFunction.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateMedianAggregateFunction.java
A hyracks-fullstack/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/FileNetworkInputChannel.java
A asterixdb/asterix-om/src/main/java/org/apache/asterix/om/typecomputer/impl/LocalMedianTypeComputer.java
M asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/BuiltinFunctions.java
A hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/JobFileState.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalSqlMedianAggregateDescriptor.java
M hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/SqlMedianAggregateDescriptor.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/functions/FunctionCollection.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalMedianAggregateDescriptor.java
M hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksJobletContext.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/scalar/ScalarMedianAggregateDescriptor.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMedianAggregateFunction.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateMedianAggregateDescriptor.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalMedianAggregateFunction.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateSqlMedianAggregateDescriptor.java
M hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Joblet.java
M hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartition.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalMedianAggregateFunction.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/SqlMedianAggregateFunction.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractLocalMedianAggregateFunction.java
M hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestJobletContext.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalSqlMedianAggregateDescriptor.java
A asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalMedianAggregateDescriptor.java
A hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionFileReaderUtil.java
M hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionManager.java
A hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionFileReader.java
29 files changed, 1,893 insertions(+), 50 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/10/18210/1
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/BuiltinFunctions.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/BuiltinFunctions.java
index 4db68c3..ca361bc 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/BuiltinFunctions.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/BuiltinFunctions.java
@@ -95,6 +95,7 @@
import org.apache.asterix.om.typecomputer.impl.InjectFailureTypeComputer;
import org.apache.asterix.om.typecomputer.impl.Int64ArrayToStringTypeComputer;
import org.apache.asterix.om.typecomputer.impl.LocalAvgTypeComputer;
+import org.apache.asterix.om.typecomputer.impl.LocalMedianTypeComputer;
import org.apache.asterix.om.typecomputer.impl.LocalSingleVarStatisticsTypeComputer;
import org.apache.asterix.om.typecomputer.impl.MinMaxAggTypeComputer;
import org.apache.asterix.om.typecomputer.impl.MissingIfTypeComputer;
@@ -551,6 +552,20 @@
new FunctionIdentifier(FunctionConstants.ASTERIX_NS, "agg-intermediate-avg", 1);
public static final FunctionIdentifier LOCAL_AVG =
new FunctionIdentifier(FunctionConstants.ASTERIX_NS, "agg-local-avg", 1);
+ public static final FunctionIdentifier MEDIAN =
+ new FunctionIdentifier(FunctionConstants.ASTERIX_NS, "agg-median", 1);
+ public static final FunctionIdentifier SQL_MEDIAN =
+ new FunctionIdentifier(FunctionConstants.ASTERIX_NS, "agg-sql-median", 1);
+ public static final FunctionIdentifier SCALAR_MEDIAN =
+ new FunctionIdentifier(FunctionConstants.ASTERIX_NS, "median", 1);
+ public static final FunctionIdentifier SCALAR_SQL_MEDIAN =
+ new FunctionIdentifier(FunctionConstants.ASTERIX_NS, "sql-median", 1);
+ public static final FunctionIdentifier LOCAL_MEDIAN =
+ new FunctionIdentifier(FunctionConstants.ASTERIX_NS, "agg-local-median", 1);
+ public static final FunctionIdentifier INTERMEDIATE_MEDIAN =
+ new FunctionIdentifier(FunctionConstants.ASTERIX_NS, "agg-intermediate-median", 1);
+ public static final FunctionIdentifier GLOBAL_MEDIAN =
+ new FunctionIdentifier(FunctionConstants.ASTERIX_NS, "agg-global-median", 1);
public static final FunctionIdentifier FIRST_ELEMENT =
new FunctionIdentifier(FunctionConstants.ASTERIX_NS, "agg-first-element", 1);
public static final FunctionIdentifier LOCAL_FIRST_ELEMENT =
@@ -1972,6 +1987,14 @@
ScalarVersionOfAggregateResultType scalarUnionMbrTypeComputer =
new ScalarVersionOfAggregateResultType(UnionMbrAggTypeComputer.INSTANCE);
+ addFunction(MEDIAN, NullableDoubleTypeComputer.INSTANCE, true);
+ addFunction(SQL_MEDIAN, NullableDoubleTypeComputer.INSTANCE, true);
+ addFunction(SCALAR_MEDIAN, NullableDoubleTypeComputer.INSTANCE, true);
+ addFunction(SCALAR_SQL_MEDIAN, NullableDoubleTypeComputer.INSTANCE, true);
+ addPrivateFunction(LOCAL_MEDIAN, LocalMedianTypeComputer.INSTANCE, true);
+ addPrivateFunction(INTERMEDIATE_MEDIAN, LocalMedianTypeComputer.INSTANCE, true);
+ addPrivateFunction(GLOBAL_MEDIAN, NullableDoubleTypeComputer.INSTANCE, true);
+
addPrivateFunction(LISTIFY, OrderedListConstructorTypeComputer.INSTANCE, true);
addFunction(SCALAR_ARRAYAGG, ScalarArrayAggTypeComputer.INSTANCE, true);
addFunction(MAX, MinMaxAggTypeComputer.INSTANCE, true);
@@ -2549,6 +2572,25 @@
static {
// Aggregate functions
+ addAgg(MEDIAN);
+ addAgg(SQL_MEDIAN);
+ addAgg(LOCAL_MEDIAN);
+ addAgg(GLOBAL_MEDIAN);
+
+ addLocalAgg(MEDIAN, LOCAL_MEDIAN);
+ addLocalAgg(SQL_MEDIAN, LOCAL_MEDIAN);
+
+ addIntermediateAgg(MEDIAN, INTERMEDIATE_MEDIAN);
+ addIntermediateAgg(SQL_MEDIAN, INTERMEDIATE_MEDIAN);
+ addIntermediateAgg(LOCAL_MEDIAN, INTERMEDIATE_MEDIAN);
+ addIntermediateAgg(GLOBAL_MEDIAN, INTERMEDIATE_MEDIAN);
+
+ addGlobalAgg(MEDIAN, GLOBAL_MEDIAN);
+ addGlobalAgg(SQL_MEDIAN, GLOBAL_MEDIAN);
+
+ addScalarAgg(MEDIAN, SCALAR_MEDIAN);
+ addScalarAgg(SQL_MEDIAN, SCALAR_SQL_MEDIAN);
+
// AVG
addAgg(AVG);
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/typecomputer/impl/LocalMedianTypeComputer.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/typecomputer/impl/LocalMedianTypeComputer.java
new file mode 100644
index 0000000..3411ef4
--- /dev/null
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/typecomputer/impl/LocalMedianTypeComputer.java
@@ -0,0 +1,44 @@
+/*
+ * 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.om.typecomputer.impl;
+
+import org.apache.asterix.om.typecomputer.base.IResultTypeComputer;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.BuiltinType;
+import org.apache.asterix.om.types.IAType;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
+import org.apache.hyracks.algebricks.core.algebra.metadata.IMetadataProvider;
+
+public class LocalMedianTypeComputer implements IResultTypeComputer {
+
+ public static final LocalMedianTypeComputer INSTANCE = new LocalMedianTypeComputer();
+
+ public static final ARecordType REC_TYPE = new ARecordType(null,
+ new String[] { "count", "handle", "address", "port" },
+ new IAType[] { BuiltinType.AINT64, BuiltinType.AINT64, BuiltinType.ASTRING, BuiltinType.AINT32 }, false);
+
+ @Override
+ public IAType computeType(ILogicalExpression expression, IVariableTypeEnvironment env,
+ IMetadataProvider<?, ?> metadataProvider) throws AlgebricksException {
+ return REC_TYPE;
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/scalar/ScalarMedianAggregateDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/scalar/ScalarMedianAggregateDescriptor.java
new file mode 100644
index 0000000..bc5d968
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/scalar/ScalarMedianAggregateDescriptor.java
@@ -0,0 +1,40 @@
+/*
+ * 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.aggregates.scalar;
+
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.runtime.aggregates.std.MedianAggregateDescriptor;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+
+public class ScalarMedianAggregateDescriptor extends AbstractScalarAggregateDescriptor {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = ScalarMedianAggregateDescriptor::new;
+
+ private ScalarMedianAggregateDescriptor() {
+ super(MedianAggregateDescriptor.FACTORY);
+ }
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.SCALAR_MEDIAN;
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractLocalMedianAggregateFunction.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractLocalMedianAggregateFunction.java
new file mode 100644
index 0000000..d04139d
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractLocalMedianAggregateFunction.java
@@ -0,0 +1,159 @@
+/*
+ * 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.aggregates.std;
+
+import java.util.List;
+
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.EnumDeserializer;
+import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.application.INCServiceContext;
+import org.apache.hyracks.api.comm.IFrameTupleAppender;
+import org.apache.hyracks.api.comm.IFrameWriter;
+import org.apache.hyracks.api.comm.NetworkAddress;
+import org.apache.hyracks.api.context.IHyracksJobletContext;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputer;
+import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.SourceLocation;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.control.nc.NodeControllerService;
+import org.apache.hyracks.control.nc.net.NetworkManager;
+import org.apache.hyracks.control.nc.partitions.JobFileState;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
+import org.apache.hyracks.dataflow.common.io.RunFileWriter;
+import org.apache.hyracks.dataflow.std.buffermanager.EnumFreeSlotPolicy;
+import org.apache.hyracks.dataflow.std.sort.Algorithm;
+import org.apache.hyracks.dataflow.std.sort.ExternalSortRunGenerator;
+import org.apache.hyracks.dataflow.std.sort.ExternalSortRunMerger;
+import org.apache.hyracks.dataflow.std.sort.ISorter;
+
+public abstract class AbstractLocalMedianAggregateFunction extends AbstractMedianAggregateFunction {
+
+ protected final ExternalSortRunGenerator runsGenerator;
+ protected final IFrameTupleAppender appender;
+ private final ArrayTupleBuilder tupleBuilder;
+ private ExternalSortRunMerger runsMerger;
+
+ public AbstractLocalMedianAggregateFunction(IScalarEvaluatorFactory[] args, IEvaluatorContext context,
+ SourceLocation sourceLoc) throws HyracksDataException {
+ super(args, context, sourceLoc);
+ appender = new FrameTupleAppender(frame);
+ tupleBuilder = new ArrayTupleBuilder(1);
+ runsGenerator = new ExternalSortRunGenerator(context.getTaskContext(), new int[] { 0 },
+ new INormalizedKeyComputerFactory[] { doubleNkComputerFactory },
+ new IBinaryComparatorFactory[] { doubleComparatorFactory }, recordDesc, Algorithm.MERGE_SORT,
+ EnumFreeSlotPolicy.LAST_FIT, 12, Integer.MAX_VALUE);
+
+ }
+
+ @Override
+ public void init() throws HyracksDataException {
+ super.init();
+ appender.reset(frame, true);
+ runsGenerator.open();
+ runsGenerator.getSorter().reset();
+ }
+
+ protected void processDataValue(IFrameTupleReference tuple) throws HyracksDataException {
+ eval.evaluate(tuple, inputVal);
+ byte[] data = inputVal.getByteArray();
+ int start = inputVal.getStartOffset();
+ ATypeTag tag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(data[start]);
+ if (ATypeHierarchy.getTypeDomain(tag) == ATypeHierarchy.Domain.NUMERIC) {
+ count++;
+ aDouble.setValue(ATypeHierarchy.getDoubleValue("median", 0, data, start));
+ tupleBuilder.reset();
+ tupleBuilder.addField(doubleSerde, aDouble);
+ FrameUtils.appendToWriter(runsGenerator, appender, tupleBuilder.getFieldEndOffsets(),
+ tupleBuilder.getByteArray(), 0, tupleBuilder.getSize());
+ }
+ }
+
+ protected void finishLocalPartial(IPointable result) throws HyracksDataException {
+ IHyracksTaskContext taskCtx = ctx.getTaskContext();
+ IHyracksJobletContext jobletCtx = taskCtx.getJobletContext();
+ INCServiceContext srvCtx = jobletCtx.getServiceContext();
+ NetworkManager netManager = ((NodeControllerService) srvCtx.getControllerService()).getNetworkManager();
+ NetworkAddress netAddress = netManager.getPublicNetworkAddress();
+ if (count == 0) {
+ setPartialResult(result, -1, netAddress.getAddress(), netAddress.getPort());
+ return;
+ }
+ if (appender.getTupleCount() > 0) {
+ appender.write(runsGenerator, true);
+ }
+ // close to sort the in-memory data or write out sorted data to run files
+ runsGenerator.close();
+ FileReference managedFile = jobletCtx.createManagedWorkspaceFile("median");
+ writeRunFile(managedFile, taskCtx);
+ long fileId = jobletCtx.nextUniqueId();
+ taskCtx.setStateObject(new JobFileState(managedFile, jobletCtx.getJobId(), fileId));
+ setPartialResult(result, fileId, netAddress.getAddress(), netAddress.getPort());
+ }
+
+ private void writeRunFile(FileReference managedFile, IHyracksTaskContext taskCtx) throws HyracksDataException {
+ RunFileWriter runFileWriter = new RunFileWriter(managedFile, taskCtx.getIoManager());
+ List<GeneratedRunFileReader> runs = runsGenerator.getRuns();
+ if (runs.isEmpty()) {
+ try {
+ runFileWriter.open();
+ ISorter sorter = runsGenerator.getSorter();
+ if (sorter.hasRemaining()) {
+ sorter.flush(runFileWriter);
+ }
+ } finally {
+ runFileWriter.close();
+ }
+ } else {
+ //TODO: if runs is = 1, no need to write out to file
+ createOrResetRunsMerger(runs);
+ IFrameWriter wrappingWriter = runsMerger.prepareFinalMergeResultWriter(runFileWriter);
+ try {
+ wrappingWriter.open();
+ runsMerger.process(wrappingWriter);
+ } finally {
+ wrappingWriter.close();
+ }
+ }
+ }
+
+ private void createOrResetRunsMerger(List<GeneratedRunFileReader> runs) {
+ if (runsMerger == null) {
+ IBinaryComparator[] comparators =
+ new IBinaryComparator[] { doubleComparatorFactory.createBinaryComparator() };
+ INormalizedKeyComputer nmkComputer = doubleNkComputerFactory.createNormalizedKeyComputer();
+ runsMerger = new ExternalSortRunMerger(ctx.getTaskContext(), runs, new int[] { 0 }, comparators,
+ nmkComputer, recordDesc, 12, Integer.MAX_VALUE);
+ } else {
+ runsMerger.reset(runs);
+ }
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMedianAggregateFunction.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMedianAggregateFunction.java
new file mode 100644
index 0000000..5130b5a
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMedianAggregateFunction.java
@@ -0,0 +1,369 @@
+/*
+ * 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.aggregates.std;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.asterix.builders.RecordBuilder;
+import org.apache.asterix.dataflow.data.nontagged.serde.ADoubleSerializerDeserializer;
+import org.apache.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
+import org.apache.asterix.dataflow.data.nontagged.serde.AInt64SerializerDeserializer;
+import org.apache.asterix.dataflow.data.nontagged.serde.ARecordSerializerDeserializer;
+import org.apache.asterix.formats.nontagged.BinaryComparatorFactoryProvider;
+import org.apache.asterix.formats.nontagged.NormalizedKeyComputerFactoryProvider;
+import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
+import org.apache.asterix.formats.nontagged.TypeTraitProvider;
+import org.apache.asterix.om.base.ADouble;
+import org.apache.asterix.om.base.AInt32;
+import org.apache.asterix.om.base.AInt64;
+import org.apache.asterix.om.base.AMutableDouble;
+import org.apache.asterix.om.base.AMutableInt32;
+import org.apache.asterix.om.base.AMutableInt64;
+import org.apache.asterix.om.base.AMutableString;
+import org.apache.asterix.om.base.AString;
+import org.apache.asterix.om.typecomputer.impl.LocalMedianTypeComputer;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.BuiltinType;
+import org.apache.asterix.om.types.EnumDeserializer;
+import org.apache.asterix.runtime.evaluators.functions.PointableHelper;
+import org.apache.asterix.runtime.exceptions.UnsupportedItemTypeException;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.application.INCServiceContext;
+import org.apache.hyracks.api.comm.IFrame;
+import org.apache.hyracks.api.comm.IFrameReader;
+import org.apache.hyracks.api.comm.NetworkAddress;
+import org.apache.hyracks.api.comm.VSizeFrame;
+import org.apache.hyracks.api.context.IHyracksJobletContext;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
+import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.SourceLocation;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.comm.channels.FileNetworkInputChannel;
+import org.apache.hyracks.control.nc.NodeControllerService;
+import org.apache.hyracks.control.nc.net.NetworkManager;
+import org.apache.hyracks.control.nc.partitions.JobFileState;
+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.comm.io.FrameTupleAccessor;
+import org.apache.hyracks.dataflow.common.data.accessors.FrameTupleReference;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+import org.apache.hyracks.dataflow.common.io.RunFileWriter;
+import org.apache.hyracks.dataflow.std.collectors.InputChannelFrameReader;
+import org.apache.hyracks.dataflow.std.sort.RunMergingFrameReader;
+import org.apache.hyracks.util.string.UTF8StringUtil;
+
+public abstract class AbstractMedianAggregateFunction extends AbstractAggregateFunction {
+
+ private static final int COUNT_FIELD_ID = 0;
+ private static final int HANDLE_FIELD_ID = 1;
+ private static final int ADDRESS_FIELD_ID = 2;
+ private static final int PORT_FIELD_ID = 3;
+
+ private final ArrayBackedValueStorage resultStorage = new ArrayBackedValueStorage();
+ @SuppressWarnings("unchecked")
+ private final ISerializerDeserializer<AString> stringSerde =
+ SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ASTRING);
+ @SuppressWarnings("unchecked")
+ private final ISerializerDeserializer<AInt64> longSerde =
+ SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AINT64);
+ @SuppressWarnings("unchecked")
+ private final ISerializerDeserializer<AInt32> intSerde =
+ SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AINT32);
+ @SuppressWarnings("unchecked")
+ protected final ISerializerDeserializer<ADouble> doubleSerde =
+ SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ADOUBLE);
+ protected final IBinaryComparatorFactory doubleComparatorFactory =
+ BinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory(ATypeTag.DOUBLE, true);
+ protected final INormalizedKeyComputerFactory doubleNkComputerFactory =
+ NormalizedKeyComputerFactoryProvider.INSTANCE.getNormalizedKeyComputerFactory(BuiltinType.ADOUBLE, true);
+ protected final RecordDescriptor recordDesc = new RecordDescriptor(new ISerializerDeserializer[] { doubleSerde },
+ new ITypeTraits[] { TypeTraitProvider.INSTANCE.getTypeTrait(BuiltinType.ADOUBLE) });
+
+ protected final AMutableString aString = new AMutableString("");
+ protected final AMutableInt64 aInt64 = new AMutableInt64(0);
+ protected final AMutableInt32 aInt32 = new AMutableInt32(0);
+ protected final AMutableDouble aDouble = new AMutableDouble(0);
+ protected final IPointable inputVal = new VoidPointable();
+ private final FrameTupleReference ftr = new FrameTupleReference();
+ private final FrameTupleAccessor fta = new FrameTupleAccessor(recordDesc);
+
+ protected final IScalarEvaluator eval;
+ protected final IEvaluatorContext ctx;
+ protected final IFrame frame;
+ protected long count;
+ private List<PartialResult> partialResults;
+ private RecordBuilder recBuilder;
+
+ public AbstractMedianAggregateFunction(IScalarEvaluatorFactory[] args, IEvaluatorContext context,
+ SourceLocation sourceLoc) throws HyracksDataException {
+ super(sourceLoc);
+ ctx = context;
+ eval = args[0].createScalarEvaluator(context);
+ frame = new VSizeFrame(context.getTaskContext());
+ }
+
+ @Override
+ public void init() throws HyracksDataException {
+ if (partialResults == null) {
+ partialResults = new ArrayList<>();
+ }
+ if (recBuilder == null) {
+ recBuilder = new RecordBuilder();
+ recBuilder.reset(LocalMedianTypeComputer.REC_TYPE);
+ }
+ count = 0;
+ partialResults.clear();
+ recBuilder.init();
+ }
+
+ protected void processPartialResults(IFrameTupleReference tuple) throws HyracksDataException {
+ eval.evaluate(tuple, inputVal);
+ byte[] serBytes = inputVal.getByteArray();
+ int offset = inputVal.getStartOffset();
+ ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serBytes[offset]);
+ if (typeTag == ATypeTag.OBJECT) {
+ long handleCount = AInt64SerializerDeserializer.getLong(serBytes,
+ ARecordSerializerDeserializer.getFieldOffsetById(serBytes, offset, COUNT_FIELD_ID, 0, false));
+ if (handleCount == 0) {
+ return;
+ }
+ count += handleCount;
+
+ long fileId = AInt64SerializerDeserializer.getLong(serBytes,
+ ARecordSerializerDeserializer.getFieldOffsetById(serBytes, offset, HANDLE_FIELD_ID, 0, false));
+
+ String address = UTF8StringUtil.toString(serBytes,
+ ARecordSerializerDeserializer.getFieldOffsetById(serBytes, offset, ADDRESS_FIELD_ID, 0, false));
+
+ int port = AInt32SerializerDeserializer.getInt(serBytes,
+ ARecordSerializerDeserializer.getFieldOffsetById(serBytes, offset, PORT_FIELD_ID, 0, false));
+
+ partialResults.add(new PartialResult(fileId, handleCount, address, port));
+ } else {
+ throw new UnsupportedItemTypeException(sourceLoc, "median", serBytes[offset]);
+ }
+ }
+
+ protected void finishPartialResult(IPointable result) throws HyracksDataException {
+ IHyracksTaskContext taskCtx = ctx.getTaskContext();
+ IHyracksJobletContext jobletCtx = taskCtx.getJobletContext();
+ INCServiceContext srvCtx = jobletCtx.getServiceContext();
+ NetworkManager netManager = ((NodeControllerService) srvCtx.getControllerService()).getNetworkManager();
+ NetworkAddress netAddress = netManager.getPublicNetworkAddress();
+ if (count == 0) {
+ setPartialResult(result, -1, netAddress.getAddress(), netAddress.getPort());
+ return;
+ }
+
+ List<IFrame> inFrames = new ArrayList<>(partialResults.size());
+ for (int i = 0; i < partialResults.size(); ++i) {
+ inFrames.add(new VSizeFrame(taskCtx));
+ }
+
+ JobId jobId = jobletCtx.getJobId();
+ List<IFrameReader> readers = new ArrayList<>(partialResults.size());
+ for (PartialResult partialResult : partialResults) {
+ IFrameReader inputChannelReader = createInputChannel(netManager, taskCtx,
+ new NetworkAddress(partialResult.address, partialResult.port), jobId.getId(), partialResult.fileId);
+ readers.add(inputChannelReader);
+ }
+
+ RunMergingFrameReader merger = new RunMergingFrameReader(taskCtx, readers, inFrames, new int[] { 0 },
+ new IBinaryComparator[] { doubleComparatorFactory.createBinaryComparator() },
+ doubleNkComputerFactory.createNormalizedKeyComputer(), recordDesc);
+
+ FileReference managedFile = jobletCtx.createManagedWorkspaceFile("median");
+ RunFileWriter runFileWriter = new RunFileWriter(managedFile, taskCtx.getIoManager());
+ merger.open();
+ runFileWriter.open();
+ try {
+ while (merger.nextFrame(frame)) {
+ runFileWriter.nextFrame(frame.getBuffer());
+ }
+ } finally {
+ runFileWriter.close();
+ merger.close();
+ }
+
+ long fileId = jobletCtx.nextUniqueId();
+ taskCtx.setStateObject(new JobFileState(managedFile, jobId, fileId));
+ setPartialResult(result, fileId, netAddress.getAddress(), netAddress.getPort());
+ }
+
+ protected void finishFinalResult(IPointable result) throws HyracksDataException {
+ if (count == 0) {
+ PointableHelper.setNull(result);
+ return;
+ }
+ try {
+ double medianVal = findMedian();
+ resultStorage.reset();
+ aDouble.setValue(medianVal);
+ doubleSerde.serialize(aDouble, resultStorage.getDataOutput());
+ result.set(resultStorage);
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+
+ private double findMedian() throws HyracksDataException {
+ RunMergingFrameReader merger = createRunsMergingFrameReader();
+ return getMedian(merger);
+ }
+
+ protected RunMergingFrameReader createRunsMergingFrameReader() throws HyracksDataException {
+ IHyracksTaskContext taskCtx = ctx.getTaskContext();
+ IHyracksJobletContext jobletCtx = taskCtx.getJobletContext();
+ INCServiceContext serviceContext = jobletCtx.getServiceContext();
+ NetworkManager netManager = ((NodeControllerService) serviceContext.getControllerService()).getNetworkManager();
+
+ List<IFrame> inFrames = new ArrayList<>(partialResults.size());
+ for (int i = 0; i < partialResults.size(); ++i) {
+ inFrames.add(new VSizeFrame(taskCtx));
+ }
+
+ List<IFrameReader> readers = new ArrayList<>(partialResults.size());
+ for (PartialResult partialResult : partialResults) {
+ IFrameReader inputChannelReader = createInputChannel(netManager, taskCtx,
+ new NetworkAddress(partialResult.address, partialResult.port), jobletCtx.getJobId().getId(),
+ partialResult.fileId);
+ readers.add(inputChannelReader);
+ }
+ return new RunMergingFrameReader(taskCtx, readers, inFrames, new int[] { 0 },
+ new IBinaryComparator[] { doubleComparatorFactory.createBinaryComparator() },
+ doubleNkComputerFactory.createNormalizedKeyComputer(), recordDesc);
+ }
+
+ private double getMedian(RunMergingFrameReader merger) throws HyracksDataException {
+ boolean isOdd = count % 2 != 0;
+ long medianPosition = isOdd ? count / 2 : (count - 1) / 2;
+ long currentMedianPosition = 0;
+ double medianVal = -1;
+ merger.open();
+ try {
+ while (merger.nextFrame(frame)) {
+ fta.reset(frame.getBuffer());
+ int tupleCount = fta.getTupleCount();
+ if (currentMedianPosition + tupleCount >= medianPosition) {
+ int firstMedian = (int) (medianPosition - currentMedianPosition);
+ ftr.reset(fta, firstMedian);
+ medianVal = ADoubleSerializerDeserializer.getDouble(ftr.getFieldData(0), ftr.getFieldStart(0) + 1);
+ if (!isOdd) {
+ if (firstMedian + 1 < tupleCount) {
+ ftr.reset(fta, firstMedian + 1);
+ } else {
+ merger.nextFrame(frame);
+ fta.reset(frame.getBuffer());
+ ftr.reset(fta, 0);
+ }
+ medianVal =
+ (ADoubleSerializerDeserializer.getDouble(ftr.getFieldData(0), ftr.getFieldStart(0) + 1)
+ + medianVal) / 2;
+ }
+ break;
+ }
+ currentMedianPosition = tupleCount - 1;
+ }
+ } finally {
+ merger.close();
+ }
+ return medianVal;
+ }
+
+ protected void setPartialResult(IPointable result, long fileId, String address, int port)
+ throws HyracksDataException {
+ try {
+ resultStorage.reset();
+ aInt64.setValue(count);
+ longSerde.serialize(aInt64, resultStorage.getDataOutput());
+ recBuilder.addField(COUNT_FIELD_ID, resultStorage);
+
+ resultStorage.reset();
+ aInt64.setValue(fileId);
+ longSerde.serialize(aInt64, resultStorage.getDataOutput());
+ recBuilder.addField(HANDLE_FIELD_ID, resultStorage);
+
+ resultStorage.reset();
+ aString.setValue(address);
+ stringSerde.serialize(aString, resultStorage.getDataOutput());
+ recBuilder.addField(ADDRESS_FIELD_ID, resultStorage);
+
+ resultStorage.reset();
+ aInt32.setValue(port);
+ intSerde.serialize(aInt32, resultStorage.getDataOutput());
+ recBuilder.addField(PORT_FIELD_ID, resultStorage);
+
+ resultStorage.reset();
+ recBuilder.write(resultStorage.getDataOutput(), true);
+ result.set(resultStorage);
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+
+ private IFrameReader createInputChannel(NetworkManager netManager, IHyracksTaskContext taskContext,
+ NetworkAddress networkAddress, long jobId, long fileId) throws HyracksDataException {
+ FileNetworkInputChannel FileNetworkInputChannel =
+ new FileNetworkInputChannel(netManager, getSocketAddress(networkAddress), jobId, fileId);
+ InputChannelFrameReader channelFrameReader = new InputChannelFrameReader(FileNetworkInputChannel);
+ FileNetworkInputChannel.registerMonitor(channelFrameReader);
+ FileNetworkInputChannel.open(taskContext);
+ return channelFrameReader;
+ }
+
+ private static SocketAddress getSocketAddress(NetworkAddress netAddress) throws HyracksDataException {
+ try {
+ return new InetSocketAddress(InetAddress.getByAddress(netAddress.lookupIpAddress()), netAddress.getPort());
+ } catch (UnknownHostException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+
+ public static class PartialResult {
+
+ long fileId;
+ String address;
+ int port;
+ long count;
+
+ PartialResult(long fileId, long count, String address, int port) {
+ this.fileId = fileId;
+ this.count = count;
+ this.address = address;
+ this.port = port;
+ }
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalMedianAggregateDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalMedianAggregateDescriptor.java
new file mode 100644
index 0000000..4f2bf09
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalMedianAggregateDescriptor.java
@@ -0,0 +1,54 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.runtime.aggregates.base.AbstractAggregateFunctionDynamicDescriptor;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluatorFactory;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public class GlobalMedianAggregateDescriptor extends AbstractAggregateFunctionDynamicDescriptor {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = GlobalMedianAggregateDescriptor::new;
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.GLOBAL_MEDIAN;
+ }
+
+ @Override
+ public IAggregateEvaluatorFactory createAggregateEvaluatorFactory(final IScalarEvaluatorFactory[] args) {
+ return new IAggregateEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IAggregateEvaluator createAggregateEvaluator(final IEvaluatorContext ctx)
+ throws HyracksDataException {
+ return new GlobalMedianAggregateFunction(args, ctx, sourceLoc);
+ }
+ };
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalMedianAggregateFunction.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalMedianAggregateFunction.java
new file mode 100644
index 0000000..bd447d5
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalMedianAggregateFunction.java
@@ -0,0 +1,51 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.SourceLocation;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+public class GlobalMedianAggregateFunction extends AbstractMedianAggregateFunction {
+
+ public GlobalMedianAggregateFunction(IScalarEvaluatorFactory[] args, IEvaluatorContext context,
+ SourceLocation sourceLoc) throws HyracksDataException {
+ super(args, context, sourceLoc);
+
+ }
+
+ @Override
+ public void step(IFrameTupleReference tuple) throws HyracksDataException {
+ processPartialResults(tuple);
+ }
+
+ @Override
+ public void finish(IPointable result) throws HyracksDataException {
+ finishFinalResult(result);
+ }
+
+ @Override
+ public void finishPartial(IPointable result) throws HyracksDataException {
+ finishPartialResult(result);
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalSqlMedianAggregateDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalSqlMedianAggregateDescriptor.java
new file mode 100644
index 0000000..5cf1152
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/GlobalSqlMedianAggregateDescriptor.java
@@ -0,0 +1,54 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.runtime.aggregates.base.AbstractAggregateFunctionDynamicDescriptor;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluatorFactory;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public class GlobalSqlMedianAggregateDescriptor extends AbstractAggregateFunctionDynamicDescriptor {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = GlobalSqlMedianAggregateDescriptor::new;
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.GLOBAL_MEDIAN;
+ }
+
+ @Override
+ public IAggregateEvaluatorFactory createAggregateEvaluatorFactory(final IScalarEvaluatorFactory[] args) {
+ return new IAggregateEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IAggregateEvaluator createAggregateEvaluator(final IEvaluatorContext ctx)
+ throws HyracksDataException {
+ return new GlobalMedianAggregateFunction(args, ctx, sourceLoc);
+ }
+ };
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateMedianAggregateDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateMedianAggregateDescriptor.java
new file mode 100644
index 0000000..2a4475a
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateMedianAggregateDescriptor.java
@@ -0,0 +1,54 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.runtime.aggregates.base.AbstractAggregateFunctionDynamicDescriptor;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluatorFactory;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public class IntermediateMedianAggregateDescriptor extends AbstractAggregateFunctionDynamicDescriptor {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = IntermediateMedianAggregateDescriptor::new;
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.INTERMEDIATE_MEDIAN;
+ }
+
+ @Override
+ public IAggregateEvaluatorFactory createAggregateEvaluatorFactory(final IScalarEvaluatorFactory[] args) {
+ return new IAggregateEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IAggregateEvaluator createAggregateEvaluator(final IEvaluatorContext ctx)
+ throws HyracksDataException {
+ return new IntermediateMedianAggregateFunction(args, ctx, sourceLoc);
+ }
+ };
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateMedianAggregateFunction.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateMedianAggregateFunction.java
new file mode 100644
index 0000000..93e9bfb
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateMedianAggregateFunction.java
@@ -0,0 +1,50 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.SourceLocation;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+public class IntermediateMedianAggregateFunction extends AbstractMedianAggregateFunction {
+
+ public IntermediateMedianAggregateFunction(IScalarEvaluatorFactory[] args, IEvaluatorContext context,
+ SourceLocation sourceLoc) throws HyracksDataException {
+ super(args, context, sourceLoc);
+ }
+
+ @Override
+ public void step(IFrameTupleReference tuple) throws HyracksDataException {
+ processPartialResults(tuple);
+ }
+
+ @Override
+ public void finish(IPointable result) throws HyracksDataException {
+ finishPartialResult(result);
+ }
+
+ @Override
+ public void finishPartial(IPointable result) throws HyracksDataException {
+ finishPartialResult(result);
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateSqlMedianAggregateDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateSqlMedianAggregateDescriptor.java
new file mode 100644
index 0000000..9d9c9d5
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/IntermediateSqlMedianAggregateDescriptor.java
@@ -0,0 +1,54 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.runtime.aggregates.base.AbstractAggregateFunctionDynamicDescriptor;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluatorFactory;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public class IntermediateSqlMedianAggregateDescriptor extends AbstractAggregateFunctionDynamicDescriptor {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = IntermediateSqlMedianAggregateDescriptor::new;
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.INTERMEDIATE_MEDIAN;
+ }
+
+ @Override
+ public IAggregateEvaluatorFactory createAggregateEvaluatorFactory(final IScalarEvaluatorFactory[] args) {
+ return new IAggregateEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IAggregateEvaluator createAggregateEvaluator(final IEvaluatorContext ctx)
+ throws HyracksDataException {
+ return new IntermediateMedianAggregateFunction(args, ctx, sourceLoc);
+ }
+ };
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalMedianAggregateDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalMedianAggregateDescriptor.java
new file mode 100644
index 0000000..58efd5d
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalMedianAggregateDescriptor.java
@@ -0,0 +1,54 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.runtime.aggregates.base.AbstractAggregateFunctionDynamicDescriptor;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluatorFactory;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public class LocalMedianAggregateDescriptor extends AbstractAggregateFunctionDynamicDescriptor {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = LocalMedianAggregateDescriptor::new;
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.LOCAL_MEDIAN;
+ }
+
+ @Override
+ public IAggregateEvaluatorFactory createAggregateEvaluatorFactory(final IScalarEvaluatorFactory[] args) {
+ return new IAggregateEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IAggregateEvaluator createAggregateEvaluator(final IEvaluatorContext ctx)
+ throws HyracksDataException {
+ return new LocalMedianAggregateFunction(args, ctx, sourceLoc);
+ }
+ };
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalMedianAggregateFunction.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalMedianAggregateFunction.java
new file mode 100644
index 0000000..1d834b0
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalMedianAggregateFunction.java
@@ -0,0 +1,50 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.SourceLocation;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+public class LocalMedianAggregateFunction extends AbstractLocalMedianAggregateFunction {
+
+ public LocalMedianAggregateFunction(IScalarEvaluatorFactory[] args, IEvaluatorContext context,
+ SourceLocation sourceLoc) throws HyracksDataException {
+ super(args, context, sourceLoc);
+ }
+
+ @Override
+ public void step(IFrameTupleReference tuple) throws HyracksDataException {
+ processDataValue(tuple);
+ }
+
+ @Override
+ public void finishPartial(IPointable result) throws HyracksDataException {
+ finishLocalPartial(result);
+ }
+
+ @Override
+ public void finish(IPointable result) throws HyracksDataException {
+ finishLocalPartial(result);
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalSqlMedianAggregateDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalSqlMedianAggregateDescriptor.java
new file mode 100644
index 0000000..1526085
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/LocalSqlMedianAggregateDescriptor.java
@@ -0,0 +1,54 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.runtime.aggregates.base.AbstractAggregateFunctionDynamicDescriptor;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluatorFactory;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public class LocalSqlMedianAggregateDescriptor extends AbstractAggregateFunctionDynamicDescriptor {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = LocalSqlMedianAggregateDescriptor::new;
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.LOCAL_MEDIAN;
+ }
+
+ @Override
+ public IAggregateEvaluatorFactory createAggregateEvaluatorFactory(final IScalarEvaluatorFactory[] args) {
+ return new IAggregateEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IAggregateEvaluator createAggregateEvaluator(final IEvaluatorContext ctx)
+ throws HyracksDataException {
+ return new LocalMedianAggregateFunction(args, ctx, sourceLoc);
+ }
+ };
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/MedianAggregateDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/MedianAggregateDescriptor.java
new file mode 100644
index 0000000..4ceb74d
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/MedianAggregateDescriptor.java
@@ -0,0 +1,55 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.runtime.aggregates.base.AbstractAggregateFunctionDynamicDescriptor;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluatorFactory;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public class MedianAggregateDescriptor extends AbstractAggregateFunctionDynamicDescriptor {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = MedianAggregateDescriptor::new;
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.MEDIAN;
+ }
+
+ @Override
+ public IAggregateEvaluatorFactory createAggregateEvaluatorFactory(final IScalarEvaluatorFactory[] args) {
+ return new IAggregateEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IAggregateEvaluator createAggregateEvaluator(final IEvaluatorContext ctx)
+ throws HyracksDataException {
+ return new MedianAggregateFunction(args, ctx, sourceLoc);
+ }
+ };
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/MedianAggregateFunction.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/MedianAggregateFunction.java
new file mode 100644
index 0000000..5d7ba6b
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/MedianAggregateFunction.java
@@ -0,0 +1,73 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
+import org.apache.asterix.om.base.ADouble;
+import org.apache.asterix.om.base.AMutableDouble;
+import org.apache.asterix.om.types.BuiltinType;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.SourceLocation;
+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;
+
+public class MedianAggregateFunction extends AbstractAggregateFunction {
+
+ private final ArrayBackedValueStorage resultStorage = new ArrayBackedValueStorage();
+ private final AMutableDouble aDouble = new AMutableDouble(0);
+ private final ISerializerDeserializer<ADouble> doubleSerde =
+ SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ADOUBLE);
+ private final IScalarEvaluator eval;
+
+ public MedianAggregateFunction(IScalarEvaluatorFactory[] args, IEvaluatorContext context, SourceLocation sourceLoc)
+ throws HyracksDataException {
+ super(sourceLoc);
+ eval = args[0].createScalarEvaluator(context);
+ }
+
+ @Override
+ public void init() throws HyracksDataException {
+
+ }
+
+ @Override
+ public void step(IFrameTupleReference tuple) throws HyracksDataException {
+
+ }
+
+ @Override
+ public void finish(IPointable result) throws HyracksDataException {
+ resultStorage.reset();
+ aDouble.setValue(2.1);
+ doubleSerde.serialize(aDouble, resultStorage.getDataOutput());
+ result.set(resultStorage);
+ }
+
+ @Override
+ public void finishPartial(IPointable result) throws HyracksDataException {
+ finish(result);
+ }
+
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/SqlMedianAggregateDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/SqlMedianAggregateDescriptor.java
new file mode 100644
index 0000000..14e737d
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/SqlMedianAggregateDescriptor.java
@@ -0,0 +1,56 @@
+/*
+ * 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.aggregates.std;
+
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.runtime.aggregates.base.AbstractAggregateFunctionDynamicDescriptor;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluator;
+import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluatorFactory;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public class SqlMedianAggregateDescriptor extends AbstractAggregateFunctionDynamicDescriptor {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final IFunctionDescriptorFactory FACTORY = SqlMedianAggregateDescriptor::new;
+
+ @Override
+ public FunctionIdentifier getIdentifier() {
+ return BuiltinFunctions.SQL_MEDIAN;
+ }
+
+ @Override
+ public IAggregateEvaluatorFactory createAggregateEvaluatorFactory(final IScalarEvaluatorFactory[] args) {
+ return new IAggregateEvaluatorFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IAggregateEvaluator createAggregateEvaluator(final IEvaluatorContext ctx)
+ throws HyracksDataException {
+ return new SqlMedianAggregateFunction(args, ctx, sourceLoc);
+ }
+ };
+ }
+
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/SqlMedianAggregateFunction.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/SqlMedianAggregateFunction.java
new file mode 100644
index 0000000..d427def
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/SqlMedianAggregateFunction.java
@@ -0,0 +1,106 @@
+/*
+ * 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.aggregates.std;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.asterix.runtime.evaluators.functions.PointableHelper;
+import org.apache.hyracks.algebricks.runtime.base.IEvaluatorContext;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.api.comm.IFrame;
+import org.apache.hyracks.api.comm.VSizeFrame;
+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.api.exceptions.SourceLocation;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
+import org.apache.hyracks.dataflow.common.io.RunFileWriter;
+import org.apache.hyracks.dataflow.std.sort.ISorter;
+import org.apache.hyracks.dataflow.std.sort.RunMergingFrameReader;
+
+public class SqlMedianAggregateFunction extends AbstractLocalMedianAggregateFunction {
+
+ public SqlMedianAggregateFunction(IScalarEvaluatorFactory[] args, IEvaluatorContext context,
+ SourceLocation sourceLoc) throws HyracksDataException {
+ super(args, context, sourceLoc);
+ }
+
+ @Override
+ public void step(IFrameTupleReference tuple) throws HyracksDataException {
+ processDataValue(tuple);
+ }
+
+ @Override
+ public void finish(IPointable result) throws HyracksDataException {
+ if (count == 0) {
+ PointableHelper.setNull(result);
+ return;
+ }
+ if (appender.getTupleCount() > 0) {
+ appender.write(runsGenerator, true);
+ }
+ // close to sort the in-memory data or write out sorted data to run files
+ runsGenerator.close();
+ super.finishFinalResult(result);
+ }
+
+ @Override
+ public void finishPartial(IPointable result) throws HyracksDataException {
+ finishLocalPartial(result);
+ }
+
+ @Override
+ protected RunMergingFrameReader createRunsMergingFrameReader() throws HyracksDataException {
+ IHyracksTaskContext taskCtx = ctx.getTaskContext();
+ List<GeneratedRunFileReader> runs = runsGenerator.getRuns();
+ List<GeneratedRunFileReader> readers;
+ if (runs.isEmpty()) {
+ FileReference managedFile = taskCtx.createManagedWorkspaceFile("median");
+ RunFileWriter runFileWriter = new RunFileWriter(managedFile, taskCtx.getIoManager());
+ try {
+ runFileWriter.open();
+ ISorter sorter = runsGenerator.getSorter();
+ if (sorter.hasRemaining()) {
+ sorter.flush(runFileWriter);
+ }
+ } finally {
+ runFileWriter.close();
+ }
+ GeneratedRunFileReader deleteOnCloseReader = runFileWriter.createDeleteOnCloseReader();
+ readers = new ArrayList<>(1);
+ readers.add(deleteOnCloseReader);
+ } else {
+ //TODO: by reference?
+ readers = runs;
+ }
+
+ List<IFrame> inFrames = new ArrayList<>(readers.size());
+ for (int i = 0; i < readers.size(); ++i) {
+ inFrames.add(new VSizeFrame(taskCtx));
+ }
+ return new RunMergingFrameReader(taskCtx, readers, inFrames, new int[] { 0 },
+ new IBinaryComparator[] { doubleComparatorFactory.createBinaryComparator() },
+ doubleNkComputerFactory.createNormalizedKeyComputer(), recordDesc);
+ }
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/functions/FunctionCollection.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/functions/FunctionCollection.java
index 4b418b6..aa99629 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/functions/FunctionCollection.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/functions/FunctionCollection.java
@@ -42,6 +42,7 @@
import org.apache.asterix.runtime.aggregates.scalar.ScalarKurtosisDistinctAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.scalar.ScalarMaxAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.scalar.ScalarMaxDistinctAggregateDescriptor;
+import org.apache.asterix.runtime.aggregates.scalar.ScalarMedianAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.scalar.ScalarMinAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.scalar.ScalarMinDistinctAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.scalar.ScalarSkewnessAggregateDescriptor;
@@ -151,11 +152,13 @@
import org.apache.asterix.runtime.aggregates.std.GlobalAvgAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.GlobalKurtosisAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.GlobalMaxAggregateDescriptor;
+import org.apache.asterix.runtime.aggregates.std.GlobalMedianAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.GlobalMinAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.GlobalSkewnessAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.GlobalSqlAvgAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.GlobalSqlKurtosisAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.GlobalSqlMaxAggregateDescriptor;
+import org.apache.asterix.runtime.aggregates.std.GlobalSqlMedianAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.GlobalSqlMinAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.GlobalSqlSkewnessAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.GlobalSqlStddevAggregateDescriptor;
@@ -173,11 +176,13 @@
import org.apache.asterix.runtime.aggregates.std.IntermediateAvgAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.IntermediateKurtosisAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.IntermediateMaxAggregateDescriptor;
+import org.apache.asterix.runtime.aggregates.std.IntermediateMedianAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.IntermediateMinAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.IntermediateSkewnessAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.IntermediateSqlAvgAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.IntermediateSqlKurtosisAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.IntermediateSqlMaxAggregateDescriptor;
+import org.apache.asterix.runtime.aggregates.std.IntermediateSqlMedianAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.IntermediateSqlMinAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.IntermediateSqlSkewnessAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.IntermediateSqlStddevAggregateDescriptor;
@@ -196,12 +201,14 @@
import org.apache.asterix.runtime.aggregates.std.LocalAvgAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalKurtosisAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalMaxAggregateDescriptor;
+import org.apache.asterix.runtime.aggregates.std.LocalMedianAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalMinAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalSamplingAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalSkewnessAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalSqlAvgAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalSqlKurtosisAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalSqlMaxAggregateDescriptor;
+import org.apache.asterix.runtime.aggregates.std.LocalSqlMedianAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalSqlMinAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalSqlSkewnessAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalSqlStddevAggregateDescriptor;
@@ -217,6 +224,7 @@
import org.apache.asterix.runtime.aggregates.std.LocalVarAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.LocalVarPopAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.MaxAggregateDescriptor;
+import org.apache.asterix.runtime.aggregates.std.MedianAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.MinAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.RangeMapAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.SkewnessAggregateDescriptor;
@@ -224,6 +232,7 @@
import org.apache.asterix.runtime.aggregates.std.SqlCountAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.SqlKurtosisAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.SqlMaxAggregateDescriptor;
+import org.apache.asterix.runtime.aggregates.std.SqlMedianAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.SqlMinAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.SqlSkewnessAggregateDescriptor;
import org.apache.asterix.runtime.aggregates.std.SqlStddevAggregateDescriptor;
@@ -683,6 +692,18 @@
fc.add(ListifyAggregateDescriptor.FACTORY);
fc.add(CountAggregateDescriptor.FACTORY);
fc.add(AvgAggregateDescriptor.FACTORY);
+
+ fc.add(MedianAggregateDescriptor.FACTORY);
+ fc.add(LocalMedianAggregateDescriptor.FACTORY);
+ fc.add(IntermediateMedianAggregateDescriptor.FACTORY);
+ fc.add(GlobalMedianAggregateDescriptor.FACTORY);
+ fc.add(ScalarMedianAggregateDescriptor.FACTORY);
+
+ fc.add(SqlMedianAggregateDescriptor.FACTORY);
+ fc.add(LocalSqlMedianAggregateDescriptor.FACTORY);
+ fc.add(IntermediateSqlMedianAggregateDescriptor.FACTORY);
+ fc.add(GlobalSqlMedianAggregateDescriptor.FACTORY);
+
fc.add(LocalAvgAggregateDescriptor.FACTORY);
fc.add(IntermediateAvgAggregateDescriptor.FACTORY);
fc.add(GlobalAvgAggregateDescriptor.FACTORY);
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksJobletContext.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksJobletContext.java
index 4d324f4..422782e 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksJobletContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksJobletContext.java
@@ -44,4 +44,6 @@
Class<?> loadClass(String className) throws HyracksException;
ClassLoader getClassLoader() throws HyracksException;
+
+ long nextUniqueId();
}
diff --git a/hyracks-fullstack/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/FileNetworkInputChannel.java b/hyracks-fullstack/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/FileNetworkInputChannel.java
new file mode 100644
index 0000000..9e18cba
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/FileNetworkInputChannel.java
@@ -0,0 +1,145 @@
+/*
+ * 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.hyracks.comm.channels;
+
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
+import java.util.Queue;
+
+import org.apache.hyracks.api.channels.IInputChannel;
+import org.apache.hyracks.api.channels.IInputChannelMonitor;
+import org.apache.hyracks.api.comm.IBufferAcceptor;
+import org.apache.hyracks.api.comm.IChannelControlBlock;
+import org.apache.hyracks.api.comm.ICloseableBufferAcceptor;
+import org.apache.hyracks.api.context.IHyracksCommonContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public class FileNetworkInputChannel implements IInputChannel {
+
+ private static final int INITIAL_MESSAGE_SIZE = 24;
+ private static final int NUM_READ_BUFFERS = 1;
+ public static final long FILE_CHANNEL_CODE = -1;
+
+ private final IChannelConnectionFactory netManager;
+ private final SocketAddress remoteAddress;
+ private final long jobId;
+ private final long fileId;
+ private final Queue<ByteBuffer> fullQueue;
+ private final int nBuffers;
+ private IChannelControlBlock ccb;
+ private IInputChannelMonitor monitor;
+ private Object attachment;
+
+ public FileNetworkInputChannel(IChannelConnectionFactory netManager, SocketAddress remoteAddress, long jobId,
+ long fileId) {
+ this.netManager = netManager;
+ this.remoteAddress = remoteAddress;
+ this.jobId = jobId;
+ this.fileId = fileId;
+ this.fullQueue = new ArrayDeque<>(NUM_READ_BUFFERS);
+ this.nBuffers = NUM_READ_BUFFERS;
+ }
+
+ @Override
+ public void registerMonitor(IInputChannelMonitor monitor) {
+ this.monitor = monitor;
+ }
+
+ @Override
+ public void setAttachment(Object attachment) {
+ this.attachment = attachment;
+ }
+
+ @Override
+ public Object getAttachment() {
+ return attachment;
+ }
+
+ @Override
+ public synchronized ByteBuffer getNextBuffer() {
+ return fullQueue.poll();
+ }
+
+ @Override
+ public void recycleBuffer(ByteBuffer buffer) {
+ buffer.clear();
+ ccb.getReadInterface().getEmptyBufferAcceptor().accept(buffer);
+ }
+
+ @Override
+ public void open(IHyracksCommonContext ctx) throws HyracksDataException {
+ try {
+ ccb = netManager.connect(remoteAddress);
+ } catch (Exception e) {
+ throw HyracksDataException.create(e);
+ }
+ ccb.getReadInterface().setFullBufferAcceptor(new ReadFullBufferAcceptor());
+ ccb.getWriteInterface().setEmptyBufferAcceptor(WriteEmptyBufferAcceptor.INSTANCE);
+ ccb.getReadInterface().setBufferFactory(new ReadBufferFactory(nBuffers, ctx), nBuffers,
+ ctx.getInitialFrameSize());
+
+ ByteBuffer writeBuffer = ByteBuffer.allocate(INITIAL_MESSAGE_SIZE);
+ writeBuffer.putLong(FILE_CHANNEL_CODE);
+ writeBuffer.putLong(jobId);
+ writeBuffer.putLong(fileId);
+ writeBuffer.flip();
+
+ ccb.getWriteInterface().getFullBufferAcceptor().accept(writeBuffer);
+ ccb.getWriteInterface().getFullBufferAcceptor().close();
+ }
+
+ @Override
+ public void close() throws HyracksDataException {
+
+ }
+
+ @Override
+ public void fail() {
+ // do nothing (covered by job lifecycle)
+ }
+
+ private class ReadFullBufferAcceptor implements ICloseableBufferAcceptor {
+ @Override
+ public void accept(ByteBuffer buffer) {
+ fullQueue.add(buffer);
+ monitor.notifyDataAvailability(FileNetworkInputChannel.this, 1);
+ }
+
+ @Override
+ public void close() {
+ monitor.notifyEndOfStream(FileNetworkInputChannel.this);
+ }
+
+ @Override
+ public void error(int ecode) {
+ monitor.notifyFailure(FileNetworkInputChannel.this, ecode);
+ }
+ }
+
+ private static class WriteEmptyBufferAcceptor implements IBufferAcceptor {
+
+ static final WriteEmptyBufferAcceptor INSTANCE = new WriteEmptyBufferAcceptor();
+
+ @Override
+ public void accept(ByteBuffer buffer) {
+ // do nothing
+ }
+ }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Joblet.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Joblet.java
index f179c36..a96dfe5 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Joblet.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Joblet.java
@@ -108,6 +108,7 @@
private final String jobStartTimeZoneId;
private final long maxWarnings;
+ private final AtomicLong uniqueIds;
public Joblet(NodeControllerService nodeController, DeploymentId deploymentId, JobId jobId,
INCServiceContext serviceCtx, ActivityClusterGraph acg,
@@ -140,6 +141,7 @@
this.jobStartTime = jobStartTime;
this.jobStartTimeZoneId = jobStartTimeZoneId;
this.maxWarnings = acg.getMaxWarnings();
+ this.uniqueIds = new AtomicLong();
}
@Override
@@ -161,6 +163,11 @@
}
@Override
+ public long nextUniqueId() {
+ return uniqueIds.getAndIncrement();
+ }
+
+ @Override
public long getJobStartTime() {
return jobStartTime;
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
index 6876618..d15cdcc 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
@@ -31,8 +31,11 @@
import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.api.network.ISocketChannelFactory;
import org.apache.hyracks.api.partitions.PartitionId;
+import org.apache.hyracks.comm.channels.FileNetworkInputChannel;
import org.apache.hyracks.comm.channels.IChannelConnectionFactory;
import org.apache.hyracks.comm.channels.NetworkOutputChannel;
+import org.apache.hyracks.control.nc.NodeControllerService;
+import org.apache.hyracks.control.nc.partitions.PartitionFileReaderUtil;
import org.apache.hyracks.control.nc.partitions.PartitionManager;
import org.apache.hyracks.net.protocols.muxdemux.ChannelControlBlock;
import org.apache.hyracks.net.protocols.muxdemux.IChannelOpenListener;
@@ -47,7 +50,7 @@
private static final int MAX_CONNECTION_ATTEMPTS = 5;
- static final int INITIAL_MESSAGE_SIZE = 20;
+ static final int INITIAL_MESSAGE_SIZE = 24;
private final PartitionManager partitionManager;
@@ -128,12 +131,13 @@
@Override
public void accept(ByteBuffer buffer) {
- PartitionId pid = readInitialMessage(buffer);
- if (LOGGER.isTraceEnabled()) {
- LOGGER.trace("Received initial partition request: " + pid + " on channel: " + ccb);
- }
noc = new NetworkOutputChannel(ccb, nBuffers);
- partitionManager.registerPartitionRequest(pid, noc);
+ long id = buffer.getLong();
+ if (id == FileNetworkInputChannel.FILE_CHANNEL_CODE) {
+ handleFileRequest(buffer);
+ } else {
+ handlePartitionRequest(buffer, id);
+ }
}
@Override
@@ -147,16 +151,37 @@
noc.abort(ecode);
}
}
+
+ private void handlePartitionRequest(ByteBuffer buffer, long jid) {
+ PartitionId pid = readInitialMessage(buffer, jid);
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Received initial partition request: " + pid + " on channel: " + ccb);
+ }
+ partitionManager.registerPartitionRequest(pid, noc);
+ }
+
+ private void handleFileRequest(ByteBuffer buffer) {
+ JobId jobId = new JobId(buffer.getLong());
+ long fileId = buffer.getLong();
+ if (partitionManager.registerFileRequest(jobId, noc)) {
+ writeFileToChannel(partitionManager.getNodeControllerService(), noc, jobId, fileId);
+ }
+ }
}
- private static PartitionId readInitialMessage(ByteBuffer buffer) {
- JobId jobId = new JobId(buffer.getLong());
+ private static PartitionId readInitialMessage(ByteBuffer buffer, long jid) {
+ JobId jobId = new JobId(jid);
ConnectorDescriptorId cdid = new ConnectorDescriptorId(buffer.getInt());
int senderIndex = buffer.getInt();
int receiverIndex = buffer.getInt();
return new PartitionId(jobId, cdid, senderIndex, receiverIndex);
}
+ private static void writeFileToChannel(NodeControllerService ncs, NetworkOutputChannel noc, JobId jobId,
+ long fileId) {
+ PartitionFileReaderUtil.writeFileToChannel(ncs, noc, jobId, fileId);
+ }
+
public MuxDemuxPerformanceCounters getPerformanceCounters() {
return md.getPerformanceCounters();
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/JobFileState.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/JobFileState.java
new file mode 100644
index 0000000..3f814cf
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/JobFileState.java
@@ -0,0 +1,92 @@
+/*
+ * 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.hyracks.control.nc.partitions;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hyracks.api.dataflow.state.IStateObject;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.job.JobId;
+
+public class JobFileState implements IStateObject {
+
+ private final FileReference fileRef;
+ private final JobId jobId;
+ private final JobFileId jobFileId;
+
+ public JobFileState(FileReference fileRef, JobId jobId, long fileId) {
+ this.fileRef = fileRef;
+ this.jobFileId = new JobFileId(fileId);
+ this.jobId = jobId;
+ }
+
+ @Override
+ public JobId getJobId() {
+ return jobId;
+ }
+
+ @Override
+ public Object getId() {
+ return jobFileId;
+ }
+
+ @Override
+ public long getMemoryOccupancy() {
+ return 0;
+ }
+
+ @Override
+ public void toBytes(DataOutput out) throws IOException {
+
+ }
+
+ @Override
+ public void fromBytes(DataInput in) throws IOException {
+
+ }
+
+ public FileReference getFileRef() {
+ return fileRef;
+ }
+
+ public static class JobFileId {
+
+ private final long fileId;
+
+ public JobFileId(long fileId) {
+ this.fileId = fileId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof JobFileId)) {
+ return false;
+ }
+ JobFileId otherFileId = (JobFileId) o;
+ return otherFileId.fileId == fileId;
+ }
+
+ @Override
+ public int hashCode() {
+ return (int) (fileId ^ (fileId >>> 32));
+ }
+ }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartition.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartition.java
index 3e36946..218e557 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartition.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartition.java
@@ -18,14 +18,11 @@
*/
package org.apache.hyracks.control.nc.partitions;
-import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
-import org.apache.hyracks.api.io.IFileHandle;
import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.api.partitions.IPartition;
@@ -60,45 +57,7 @@
@Override
public void writeTo(final IFrameWriter writer) {
- executor.execute(new Runnable() {
- @Override
- public void run() {
- try {
- if (partitionFile == null) {
- writer.open();
- writer.close();
- return;
- }
- IFileHandle fh = ioManager.open(partitionFile, IIOManager.FileReadWriteMode.READ_ONLY,
- IIOManager.FileSyncMode.METADATA_ASYNC_DATA_ASYNC);
- try {
- writer.open();
- try {
- long offset = 0;
- ByteBuffer buffer = ctx.allocateFrame();
- while (true) {
- buffer.clear();
- long size = ioManager.syncRead(fh, offset, buffer);
- if (size < 0) {
- break;
- } else if (size < buffer.capacity()) {
- throw new HyracksDataException("Premature end of file");
- }
- offset += size;
- buffer.flip();
- writer.nextFrame(buffer);
- }
- } finally {
- writer.close();
- }
- } finally {
- ioManager.close(fh);
- }
- } catch (HyracksDataException e) {
- throw new RuntimeException(e);
- }
- }
- });
+ executor.execute(new PartitionFileReader(ctx, partitionFile, ioManager, writer));
}
@Override
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionFileReader.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionFileReader.java
new file mode 100644
index 0000000..a0d8dcd
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionFileReader.java
@@ -0,0 +1,82 @@
+/*
+ * 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.hyracks.control.nc.partitions;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hyracks.api.comm.IFrameWriter;
+import org.apache.hyracks.api.context.IHyracksCommonContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.io.IFileHandle;
+import org.apache.hyracks.api.io.IIOManager;
+
+public class PartitionFileReader implements Runnable {
+
+ private final IHyracksCommonContext ctx;
+ private final FileReference partitionFile;
+ private final IIOManager ioManager;
+ private final IFrameWriter writer;
+
+ public PartitionFileReader(IHyracksCommonContext ctx, FileReference partitionFile, IIOManager ioManager,
+ IFrameWriter writer) {
+ this.ctx = ctx;
+ this.partitionFile = partitionFile;
+ this.ioManager = ioManager;
+ this.writer = writer;
+ }
+
+ @Override
+ public void run() {
+ try {
+ if (partitionFile == null) {
+ writer.open();
+ writer.close();
+ return;
+ }
+ IFileHandle fh = ioManager.open(partitionFile, IIOManager.FileReadWriteMode.READ_ONLY,
+ IIOManager.FileSyncMode.METADATA_ASYNC_DATA_ASYNC);
+ try {
+ writer.open();
+ try {
+ long offset = 0;
+ ByteBuffer buffer = ctx.allocateFrame();
+ while (true) {
+ buffer.clear();
+ long size = ioManager.syncRead(fh, offset, buffer);
+ if (size < 0) {
+ break;
+ } else if (size < buffer.capacity()) {
+ throw new HyracksDataException("Premature end of file");
+ }
+ offset += size;
+ buffer.flip();
+ writer.nextFrame(buffer);
+ }
+ } finally {
+ writer.close();
+ }
+ } finally {
+ ioManager.close(fh);
+ }
+ } catch (HyracksDataException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionFileReaderUtil.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionFileReaderUtil.java
new file mode 100644
index 0000000..177d7cc
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionFileReaderUtil.java
@@ -0,0 +1,52 @@
+/*
+ * 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.hyracks.control.nc.partitions;
+
+import java.util.concurrent.ExecutorService;
+
+import org.apache.hyracks.api.dataflow.state.IStateObject;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.comm.channels.NetworkOutputChannel;
+import org.apache.hyracks.control.nc.Joblet;
+import org.apache.hyracks.control.nc.NodeControllerService;
+import org.apache.hyracks.net.protocols.muxdemux.AbstractChannelWriteInterface;
+
+public class PartitionFileReaderUtil {
+
+ private PartitionFileReaderUtil() {
+ }
+
+ public static void writeFileToChannel(NodeControllerService ncs, NetworkOutputChannel noc, JobId jobId,
+ long fileId) {
+ Joblet joblet = ncs.getJobletMap().get(jobId);
+ if (joblet == null) {
+ noc.abort(AbstractChannelWriteInterface.REMOTE_ERROR_CODE);
+ return;
+ }
+ IStateObject stateObject = joblet.getEnvironment().getStateObject(new JobFileState.JobFileId(fileId));
+ if (!(stateObject instanceof JobFileState)) {
+ noc.abort(AbstractChannelWriteInterface.REMOTE_ERROR_CODE);
+ return;
+ }
+ JobFileState fileState = (JobFileState) stateObject;
+ ExecutorService executor = ncs.getExecutor();
+ noc.setFrameSize(joblet.getInitialFrameSize());
+ executor.execute(new PartitionFileReader(joblet, fileState.getFileRef(), ncs.getIoManager(), noc));
+ }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionManager.java
index c082b71..927dd7c 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PartitionManager.java
@@ -58,6 +58,8 @@
private final Map<PartitionId, NetworkOutputChannel> partitionRequests = new HashMap<>();
+ private final Map<JobId, List<NetworkOutputChannel>> fileRequests = new HashMap<>();
+
private final Cache<JobId, JobId> failedJobsCache;
public PartitionManager(NodeControllerService ncs) {
@@ -122,10 +124,24 @@
}
}
+ public synchronized boolean registerFileRequest(JobId jid, NetworkOutputChannel noc) {
+ if (failedJobsCache.getIfPresent(jid) != null) {
+ noc.abort(AbstractChannelWriteInterface.REMOTE_ERROR_CODE);
+ return false;
+ }
+ List<NetworkOutputChannel> netOutChannels = fileRequests.computeIfAbsent(jid, k -> new ArrayList<>());
+ netOutChannels.add(noc);
+ return true;
+ }
+
public IWorkspaceFileFactory getFileFactory() {
return fileFactory;
}
+ public NodeControllerService getNodeControllerService() {
+ return ncs;
+ }
+
public void close() {
deallocatableRegistry.close();
}
@@ -175,6 +191,10 @@
}
private List<NetworkOutputChannel> removePendingRequests(JobId jobId, JobStatus status) {
+ List<NetworkOutputChannel> jobFileRequests = null;
+ if (!fileRequests.isEmpty()) {
+ jobFileRequests = fileRequests.remove(jobId);
+ }
if (status != JobStatus.FAILURE) {
return Collections.emptyList();
}
@@ -189,6 +209,9 @@
requestsIterator.remove();
}
}
+ if (jobFileRequests != null && !jobFileRequests.isEmpty()) {
+ pendingRequests.addAll(jobFileRequests);
+ }
return pendingRequests;
}
}
diff --git a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestJobletContext.java b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestJobletContext.java
index 46e3eec..bf4cbd0 100644
--- a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestJobletContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestJobletContext.java
@@ -20,6 +20,7 @@
import java.nio.ByteBuffer;
import java.time.ZoneId;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.hyracks.api.application.INCServiceContext;
import org.apache.hyracks.api.context.IHyracksJobletContext;
@@ -42,6 +43,7 @@
private final WorkspaceFileFactory fileFactory;
private final long jobStartTime;
private final String jobStartTimeZoneId;
+ private final AtomicLong ids;
TestJobletContext(int frameSize, INCServiceContext serviceContext, JobId jobId) throws HyracksException {
this.serviceContext = serviceContext;
@@ -50,6 +52,7 @@
this.frameManger = new FrameManager(frameSize);
this.jobStartTime = System.currentTimeMillis();
this.jobStartTimeZoneId = ZoneId.systemDefault().getId();
+ this.ids = new AtomicLong();
}
@Override
@@ -151,4 +154,8 @@
return this.getClass().getClassLoader();
}
+ @Override
+ public long nextUniqueId() {
+ return ids.getAndIncrement();
+ }
}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/18210
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: neo
Gerrit-Change-Id: Id8d03f42d54b5ed4cf316c31f0b4cce9dd7c1dc0
Gerrit-Change-Number: 18210
Gerrit-PatchSet: 1
Gerrit-Owner: Ali Alsuliman <al...@gmail.com>
Gerrit-MessageType: newchange