From 4c399b361a9427bc9b7009381c242652eac94507 Mon Sep 17 00:00:00 2001 From: Jannik Lindemann Date: Fri, 31 Jul 2026 18:47:14 +0200 Subject: [PATCH] [OOC] Add NaryJoinOOCPrimitive and Generic Join --- .../ooc/TernaryOOCInstruction.java | 32 ++ .../spark/data/IndexedMatrixValue.java | 6 + .../sysds/runtime/ooc/cache/OOCFuture.java | 52 +++ .../runtime/ooc/cache/io/SpillableObject.java | 1 + .../runtime/ooc/cache/packed/PackedBlock.java | 5 + .../ooc/primitives/JoinOOCPrimitive.java | 153 +++++---- .../ooc/primitives/NaryJoinOOCPrimitive.java | 307 ++++++++++++++++++ .../sysds/runtime/ooc/store/StateTable.java | 10 + .../runtime/ooc/util/OOCInstructionUtils.java | 37 ++- .../runtime/ooc/util/StateTableUtils.java | 78 +++-- .../test/component/ooc/OOCPrimitiveTest.java | 37 +++ .../test/functions/ooc/TernaryMatrixTest.java | 105 ++++++ .../scripts/functions/ooc/TernaryMatrix.dml | 36 ++ 13 files changed, 768 insertions(+), 91 deletions(-) create mode 100644 src/main/java/org/apache/sysds/runtime/ooc/primitives/NaryJoinOOCPrimitive.java create mode 100644 src/test/java/org/apache/sysds/test/functions/ooc/TernaryMatrixTest.java create mode 100644 src/test/scripts/functions/ooc/TernaryMatrix.dml diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/TernaryOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TernaryOOCInstruction.java index 7b91b16d237..b0c8aaf27a5 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/TernaryOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TernaryOOCInstruction.java @@ -26,6 +26,11 @@ import org.apache.sysds.runtime.controlprogram.caching.MatrixObject; import org.apache.sysds.runtime.controlprogram.context.ExecutionContext; import org.apache.sysds.runtime.functionobjects.IfElse; +import org.apache.sysds.runtime.functionobjects.Minus; +import org.apache.sysds.runtime.functionobjects.MinusMultiply; +import org.apache.sysds.runtime.functionobjects.Multiply; +import org.apache.sysds.runtime.functionobjects.Plus; +import org.apache.sysds.runtime.functionobjects.PlusMultiply; import org.apache.sysds.runtime.instructions.InstructionUtils; import org.apache.sysds.runtime.instructions.cp.CPOperand; import org.apache.sysds.runtime.instructions.cp.ScalarObject; @@ -33,6 +38,7 @@ import org.apache.sysds.runtime.instructions.cp.StringObject; import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue; import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.matrix.operators.BinaryOperator; import org.apache.sysds.runtime.matrix.operators.Operator; import org.apache.sysds.runtime.matrix.operators.TernaryOperator; import org.apache.sysds.runtime.ooc.util.OOCInstructionUtils; @@ -158,6 +164,32 @@ private void processThreeMatrixInstruction(ExecutionContext ec) { OOCStream qOut = createWritableStream(); ec.getMatrixObject(output).setStreamHandle(qOut); + if(m1.getDataCharacteristics().dimsKnown() && m2.getDataCharacteristics().dimsKnown() && + m3.getDataCharacteristics().dimsKnown()) { + TernaryOperator operator = (TernaryOperator) _optr; + if(operator.fn instanceof PlusMultiply || operator.fn instanceof MinusMultiply) { + OOCStream product = createWritableStream(); + BinaryOperator multiply = new BinaryOperator(Multiply.getMultiplyFnObject()); + BinaryOperator combine = operator.fn instanceof PlusMultiply ? new BinaryOperator( + Plus.getPlusFnObject()) : new BinaryOperator(Minus.getMinusFnObject()); + OOCInstructionUtils.equiJoin(m2.getStreamable(), m3.getStreamable(), product, + (left, right) -> left.binaryOperations(multiply, right, new MatrixBlock()), getContext()); + OOCInstructionUtils.equiJoin(m1.getStreamable(), product, qOut, + (left, right) -> left.binaryOperations(combine, right, new MatrixBlock()), getContext()); + return; + } + if(operator.fn instanceof IfElse) { + OOCInstructionUtils.naryEquiJoin(List.of(m1.getStreamable(), m2.getStreamable(), m3.getStreamable()), + qOut, + blocks -> new IndexedMatrixValue(blocks.get(0).getIndexes(), + ((MatrixBlock) blocks.get(0).getValue()).ternaryOperations(operator, + (MatrixBlock) blocks.get(1).getValue(), (MatrixBlock) blocks.get(2).getValue(), + new MatrixBlock())), + getContext()); + return; + } + } + List> streams = List.of( m1.getStreamHandle(), m2.getStreamHandle(), m3.getStreamHandle()); diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/data/IndexedMatrixValue.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/data/IndexedMatrixValue.java index bd96bbb614f..2f83caa5526 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/spark/data/IndexedMatrixValue.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/data/IndexedMatrixValue.java @@ -97,6 +97,12 @@ public boolean tryWrite(DataOutput dataOutput) throws IOException { return true; } + @Override + public long size() { + MatrixBlock block = (MatrixBlock) _value; + return Math.max(block.getExactSerializedSize(), block.getInMemorySize()); + } + @Override public void discard() { _value = null; diff --git a/src/main/java/org/apache/sysds/runtime/ooc/cache/OOCFuture.java b/src/main/java/org/apache/sysds/runtime/ooc/cache/OOCFuture.java index d6796e35a87..0e9504ce08f 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/cache/OOCFuture.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/cache/OOCFuture.java @@ -19,10 +19,16 @@ package org.apache.sysds.runtime.ooc.cache; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; import java.util.concurrent.CompletionException; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.function.Function; @@ -51,6 +57,52 @@ public static OOCFuture failed(Throwable error) { return future; } + public static OOCFuture> allOf(List> futures, + Consumer failureCleanup) { + Objects.requireNonNull(futures); + Objects.requireNonNull(failureCleanup); + if(futures.isEmpty()) + return completed(List.of()); + OOCFuture> result = new OOCFuture<>(); + Object[] values = new Object[futures.size()]; + AtomicInteger remaining = new AtomicInteger(futures.size()); + AtomicReference firstError = new AtomicReference<>(); + for(int i = 0; i < futures.size(); i++) { + int index = i; + futures.get(i).whenComplete((value, error) -> { + values[index] = value; + if(error != null) + firstError.compareAndSet(null, error); + if(remaining.decrementAndGet() != 0) + return; + Throwable failure = firstError.get(); + List completed = new ArrayList<>(values.length); + for(Object item : values) { + @SuppressWarnings("unchecked") + T typed = (T) item; + completed.add(typed); + } + if(failure == null) { + result.complete(Collections.unmodifiableList(completed)); + return; + } + for(T item : completed) { + if(item == null) + continue; + try { + failureCleanup.accept(item); + } + catch(Throwable cleanupError) { + if(cleanupError != failure) + failure.addSuppressed(cleanupError); + } + } + result.completeExceptionally(failure); + }); + } + return result; + } + public boolean complete(T value) { return finish(value, null); } diff --git a/src/main/java/org/apache/sysds/runtime/ooc/cache/io/SpillableObject.java b/src/main/java/org/apache/sysds/runtime/ooc/cache/io/SpillableObject.java index 434f70601d6..a93b1d18f2a 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/cache/io/SpillableObject.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/cache/io/SpillableObject.java @@ -26,6 +26,7 @@ public interface SpillableObject { boolean tryWrite(DataOutput out) throws IOException; void read(DataInput in) throws IOException; + long size(); default void discard() { } diff --git a/src/main/java/org/apache/sysds/runtime/ooc/cache/packed/PackedBlock.java b/src/main/java/org/apache/sysds/runtime/ooc/cache/packed/PackedBlock.java index 5727e2eea63..5cbb8976c33 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/cache/packed/PackedBlock.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/cache/packed/PackedBlock.java @@ -57,6 +57,11 @@ public boolean tryWrite(DataOutput out) throws IOException { return true; } + @Override + public long size() { + return totalSize; + } + @Override public void read(DataInput in) throws IOException { int count = in.readInt(); diff --git a/src/main/java/org/apache/sysds/runtime/ooc/primitives/JoinOOCPrimitive.java b/src/main/java/org/apache/sysds/runtime/ooc/primitives/JoinOOCPrimitive.java index 0ec802b181c..bcc74cf21a7 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/primitives/JoinOOCPrimitive.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/primitives/JoinOOCPrimitive.java @@ -22,16 +22,18 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiFunction; +import java.util.function.ToIntFunction; +import java.util.function.ToLongFunction; import org.apache.sysds.runtime.DMLRuntimeException; import org.apache.sysds.runtime.instructions.ooc.CachingStream; import org.apache.sysds.runtime.instructions.ooc.OOCStream; import org.apache.sysds.runtime.instructions.ooc.OOCStreamable; import org.apache.sysds.runtime.instructions.ooc.SubscribableTaskQueue; -import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue; -import org.apache.sysds.runtime.matrix.data.MatrixBlock; -import org.apache.sysds.runtime.ooc.cache.OOCFuture; import org.apache.sysds.runtime.ooc.cache.OOCCacheManager; +import org.apache.sysds.runtime.ooc.cache.OOCFuture; +import org.apache.sysds.runtime.ooc.cache.io.SpillableObject; +import org.apache.sysds.runtime.ooc.memory.InMemoryQueueCallback; import org.apache.sysds.runtime.ooc.memory.ReservationBudget; import org.apache.sysds.runtime.ooc.planning.OOCAccessPattern; import org.apache.sysds.runtime.ooc.store.StateTable; @@ -40,24 +42,29 @@ import org.apache.sysds.runtime.ooc.util.OOCUtils; import org.apache.sysds.runtime.ooc.util.StateTableUtils; -public class JoinOOCPrimitive extends OOCPrimitive { - private final OOCStreamable _left; - private final OOCStreamable _right; - private final OOCStreamable _output; - private final BiFunction _operation; +public class JoinOOCPrimitive extends OOCPrimitive { + private final OOCStreamable _output; + private final ToIntFunction _leftKey; + private final ToIntFunction _rightKey; + private final ToLongFunction _outputSize; + private final BiFunction _operation; + private final long _taskBytes; private final AtomicInteger _pending = new AtomicInteger(1); private final AtomicInteger _unmatched = new AtomicInteger(); private final CompletableFuture _pendingCompletion = new CompletableFuture<>(); - private StateTable _table; + private StateTable _table; + private OOCStream _outputStream; - public JoinOOCPrimitive(OOCStreamable left, OOCStreamable right, - OOCStreamable output, BiFunction operation, - StreamContext context) { + public JoinOOCPrimitive(OOCStreamable left, OOCStreamable right, OOCStreamable output, + ToIntFunction leftKey, ToIntFunction rightKey, ToLongFunction outputSize, + BiFunction operation, long taskBytes, StreamContext context) { super(context, left, right); - _left = left; - _right = right; _output = output; + _leftKey = leftKey; + _rightKey = rightKey; + _outputSize = outputSize; _operation = operation; + _taskBytes = taskBytes; } @Override @@ -80,44 +87,34 @@ protected void requestPatternInternal(OOCAccessPattern accessPattern) { @Override protected void startExecution() { - OOCStream left = getInputReadStream(0); - OOCStream right = getInputReadStream(1); + OOCStream left = getInputReadStream(0); + OOCStream right = getInputReadStream(1); _table = new StateTable<>(OOCCacheManager.getGlobalCache(), CachingStream._streamSeq.getNextID()); - OOCStream output = _output.getWriteStream(); + _outputStream = _output.getWriteStream(); OOCStream matches = new SubscribableTaskQueue<>(); - long inputBytes = Math.max(OOCUtils.estimateOutputTileBytes(_left.getDataCharacteristics()), - OOCUtils.estimateOutputTileBytes(_right.getDataCharacteristics())); - long outputBytes = OOCUtils.estimateOutputTileBytes(_output.getDataCharacteristics()); - long taskBytes = outputBytes + 2 * inputBytes; - - getContext().addOutStream(output); - CompletableFuture processing = OOCInstructionUtils.submitCloseableOOCTasks(matches, (JoinWork work) -> { - IndexedMatrixValue mleft = work._left.get(); - IndexedMatrixValue mright = work._right.get(); - OOCUtils.enqueueExact(output, new IndexedMatrixValue(mleft.getIndexes(), - _operation.apply((MatrixBlock) mleft.getValue(), (MatrixBlock) mright.getValue())), work._budget); - }, getContext()); + + getContext().addOutStream(_outputStream); + CompletableFuture processing = OOCInstructionUtils.submitCloseableOOCTasks(matches, this::process, + getContext()); CompletableFuture.allOf(processing, _pendingCompletion).thenRun(() -> { try { _table.close(); onComplete(); } finally { - output.closeInput(); + _outputStream.closeInput(); } }); - OOCInstructionUtils.submitOOCTask(() -> drive(left, right, matches, taskBytes), - new StreamContext().addOutStream(output)); + OOCInstructionUtils.submitOOCTask(() -> drive(left, right, matches), + new StreamContext().addOutStream(_outputStream)); } - private void drive(OOCStream leftInput, OOCStream rightInput, - OOCStream matches, long taskBytes) { - long cols = _right.getDataCharacteristics().getNumColBlocks(); + private void drive(OOCStream leftInput, OOCStream rightInput, OOCStream matches) { try { while(true) { - OOCStream.QueueCallback left = leftInput.dequeueCB(); - OOCStream.QueueCallback right = rightInput.dequeueCB(); + OOCStream.QueueCallback left = leftInput.dequeueCB(); + OOCStream.QueueCallback right = rightInput.dequeueCB(); boolean leftEos = left == null || left.isEos(); boolean rightEos = right == null || right.isEos(); if(leftEos || rightEos) { @@ -129,8 +126,8 @@ private void drive(OOCStream leftInput, OOCStream leftInput, OOCStream callback, boolean left, long cols, long taskBytes, + @SuppressWarnings("unchecked") + private void accept(OOCStream.QueueCallback callback, boolean left, int key, OOCStream matches) { - if(callback == null) - return; - OOCStream.QueueCallback owned = null; ReservationBudget budget = null; boolean pending = false; + boolean handedOff = false; try { - owned = callback.keepOpen(); - callback.close(); - callback = null; - budget = OOCUtils.reserveBudget(_allowance, taskBytes); - IndexedMatrixValue value = owned.get(); - long row = value.getIndexes().getRowIndex() - 1; - long col = value.getIndexes().getColumnIndex() - 1; - int slot = Math.toIntExact(row * cols + col); + budget = OOCUtils.reserveBudget(_allowance, _taskBytes); _pending.incrementAndGet(); pending = true; - OOCFuture future = StateTableUtils.putOrTake(_table, slot, owned, budget); - owned = null; + OOCFuture> future = StateTableUtils.putOrTake(_table, key, + (OOCStream.QueueCallback) callback, budget); + handedOff = true; ReservationBudget pendingBudget = budget; budget = null; future.whenComplete((match, error) -> matchReady(match, left, pendingBudget, error, matches)); pending = false; } finally { + if(!handedOff) + callback.close(); if(pending) completePending(matches); - if(callback != null) - callback.close(); - if(owned != null) - owned.close(); if(budget != null) budget.close(); } } - private void matchReady(StateTableUtils.Match match, boolean left, ReservationBudget budget, Throwable error, - OOCStream matches) { + private void matchReady(StateTableUtils.Match match, boolean incomingLeft, + ReservationBudget budget, Throwable error, OOCStream matches) { JoinWork work = null; try { if(error != null) @@ -190,8 +178,7 @@ private void matchReady(StateTableUtils.Match match, boolean left, ReservationBu return; } _unmatched.decrementAndGet(); - work = left ? new JoinWork(match.left(), match.right(), budget) : new JoinWork(match.right(), match.left(), - budget); + work = new JoinWork(match.left(), match.right(), incomingLeft, budget); match = null; budget = null; matches.enqueue(work); @@ -213,6 +200,26 @@ private void matchReady(StateTableUtils.Match match, boolean left, ReservationBu } } + @SuppressWarnings("unchecked") + private void process(JoinWork work) { + SpillableObject incoming = work._incoming.get(); + SpillableObject existing = work._existing.get(); + L left = (L) (work._incomingLeft ? incoming : existing); + R right = (R) (work._incomingLeft ? existing : incoming); + O value = _operation.apply(left, right); + long bytes = _outputSize.applyAsLong(value); + work._budget.reserveBlocking(bytes); + OOCStream.QueueCallback callback = new InMemoryQueueCallback<>(value, null, work._budget, bytes); + try { + _outputStream.enqueue(callback); + callback = null; + } + finally { + if(callback != null) + callback.close(); + } + } + private void completePending(OOCStream matches) { if(_pending.decrementAndGet() != 0) return; @@ -233,22 +240,28 @@ private void completePending(OOCStream matches) { } } - private static final class JoinWork implements AutoCloseable { - private final OOCStream.QueueCallback _left; - private final OOCStream.QueueCallback _right; + private final class JoinWork implements AutoCloseable { + private final OOCStream.QueueCallback _incoming; + private final OOCStream.QueueCallback _existing; + private final boolean _incomingLeft; private final ReservationBudget _budget; - private JoinWork(OOCStream.QueueCallback left, - OOCStream.QueueCallback right, ReservationBudget budget) { - _left = left; - _right = right; + private JoinWork(OOCStream.QueueCallback incoming, + OOCStream.QueueCallback existing, boolean incomingLeft, ReservationBudget budget) { + _incoming = incoming; + _existing = existing; + _incomingLeft = incomingLeft; _budget = budget; } @Override public void close() { - try(_left; _right; _budget) { - // Release + try { + _incoming.close(); + _existing.close(); + } + finally { + _budget.close(); } } } diff --git a/src/main/java/org/apache/sysds/runtime/ooc/primitives/NaryJoinOOCPrimitive.java b/src/main/java/org/apache/sysds/runtime/ooc/primitives/NaryJoinOOCPrimitive.java new file mode 100644 index 00000000000..f239641f36e --- /dev/null +++ b/src/main/java/org/apache/sysds/runtime/ooc/primitives/NaryJoinOOCPrimitive.java @@ -0,0 +1,307 @@ +/* + * 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.sysds.runtime.ooc.primitives; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.function.ToIntFunction; +import java.util.function.ToLongFunction; + +import org.apache.sysds.runtime.DMLRuntimeException; +import org.apache.sysds.runtime.instructions.ooc.OOCStream; +import org.apache.sysds.runtime.instructions.ooc.OOCStreamable; +import org.apache.sysds.runtime.instructions.ooc.SubscribableTaskQueue; +import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue; +import org.apache.sysds.runtime.ooc.cache.OOCFuture; +import org.apache.sysds.runtime.ooc.memory.InMemoryQueueCallback; +import org.apache.sysds.runtime.ooc.memory.ReservationBudget; +import org.apache.sysds.runtime.ooc.planning.OOCAccessPattern; +import org.apache.sysds.runtime.ooc.store.StateTable; +import org.apache.sysds.runtime.ooc.store.StoreLease; +import org.apache.sysds.runtime.ooc.stream.StreamContext; +import org.apache.sysds.runtime.ooc.util.OOCInstructionUtils; +import org.apache.sysds.runtime.ooc.util.OOCUtils; +import org.apache.sysds.runtime.ooc.util.StateTableUtils; + +public final class NaryJoinOOCPrimitive extends OOCPrimitive { + private final List> _inputs; + private final OOCStreamable _output; + private final ToIntFunction _key; + private final ToLongFunction _size; + private final Function, IndexedMatrixValue> _operation; + private final long _storeTaskBytes; + private final long _joinTaskBytes; + private final AtomicInteger _active = new AtomicInteger(1); + private final CompletableFuture _activeCompletion = new CompletableFuture<>(); + private StateTable _table; + private OOCStream _ready; + private OOCStream _outputStream; + + public NaryJoinOOCPrimitive(List> inputs, + OOCStreamable output, ToIntFunction key, + ToLongFunction size, Function, IndexedMatrixValue> operation, + long storeTaskBytes, long joinTaskBytes, StreamContext context) { + super(context, inputs.toArray(OOCStreamable[]::new)); + if(inputs.size() < 2) + throw new IllegalArgumentException("N-ary join requires at least two inputs."); + _inputs = inputs; + _output = output; + _key = key; + _size = size; + _operation = operation; + _storeTaskBytes = storeTaskBytes; + _joinTaskBytes = joinTaskBytes; + } + + @Override + protected void inferPatternsInternal() { + _pattern = OOCAccessPattern.ANY; + for(OOCPrimitive child : getChildren()) + _pattern = _pattern.fused(child.getAccessPattern()); + if(_pattern.isPlannable() && _pattern != OOCAccessPattern.ANY) + for(OOCPrimitive child : getChildren()) + child.requestPattern(_pattern); + inferParentPatterns(); + } + + @Override + protected void requestPatternInternal(OOCAccessPattern accessPattern) { + _pattern = accessPattern; + for(OOCPrimitive child : getChildren()) + child.requestPattern(accessPattern); + } + + @Override + protected void startExecution() { + List> inputs = new ArrayList<>(_inputs.size()); + for(int i = 0; i < _inputs.size(); i++) + inputs.add(getInputReadStream(i)); + + int groups = (int) OOCUtils.getNumBlocks(inputs.get(0).getDataCharacteristics()); + _table = new StateTable<>(groups * inputs.size()); + _outputStream = _output.getWriteStream(); + _ready = new SubscribableTaskQueue<>(); + getContext().addOutStream(_outputStream, _ready); + CompletableFuture processing = OOCInstructionUtils.submitCloseableOOCTasks(_ready, this::process, + getContext()); + CompletableFuture.allOf(processing, _activeCompletion).whenComplete((ignored, error) -> { + if(error != null) + fail(error); + try { + _outputStream.closeInput(); + } + catch(Throwable failure) { + fail(failure); + } + finally { + onComplete(); + } + }); + OOCInstructionUtils.submitOOCTask(() -> drive(inputs), new StreamContext().addOutStream(_outputStream)); + } + + private void drive(List> inputs) { + try { + byte[] groupCtr = new byte[(int) OOCUtils.getNumBlocks(inputs.get(0).getDataCharacteristics())]; + int n = inputs.size(); + int unmatchedGroups = 0; + while(true) { + List> callbacks = new ArrayList<>(n); + try { + int eos = 0; + for(OOCStream input : inputs) { + OOCStream.QueueCallback callback = input.dequeueCB(); + callbacks.add(callback); + if(callback == null || callback.isEos()) + eos++; + } + if(eos != 0) { + if(eos != n) + throw new DMLRuntimeException("Join inputs contain a different number of blocks"); + if(unmatchedGroups != 0) + throw new DMLRuntimeException("Join inputs contain unmatched blocks"); + break; + } + + for(int i = 0; i < n; i++) { + OOCStream.QueueCallback callback = callbacks.get(i); + int group = _key.applyAsInt(callback.get()); + int count = ++groupCtr[group]; + if(count == 1) + unmatchedGroups++; + if(count == n) { + unmatchedGroups--; + onJoinGroupAvailable(group, callback, i, n); + } + else { + ReservationBudget budget = OOCUtils.reserveBudget(_allowance, _storeTaskBytes); + try { + StateTableUtils.put(_table, group * n + i, callback, budget); + } + finally { + budget.close(); + } + } + } + } + finally { + for(OOCStream.QueueCallback callback : callbacks) + if(callback != null) + callback.close(); + } + } + } + catch(Throwable failure) { + fail(failure); + throw DMLRuntimeException.of(failure); + } + finally { + completeActive(); + } + } + + private void onJoinGroupAvailable(int group, OOCStream.QueueCallback callback, + int callbackIndex, int n) { + ReservationBudget budget = null; + OOCStream.QueueCallback anchor = null; + boolean active = false; + try { + budget = OOCUtils.reserveBudget(_allowance, _joinTaskBytes); + anchor = callback.keepOpen(); + _active.incrementAndGet(); + active = true; + List>> futures = new ArrayList<>(n - 1); + try { + for(int i = 0; i < n; i++) + if(i != callbackIndex) + futures.add(_table.take(group * n + i, budget).map(lease -> { + if(lease == null) + throw new DMLRuntimeException("Join input block is missing"); + return lease; + })); + } + catch(Throwable failure) { + futures.add(OOCFuture.failed(failure)); + } + OOCFuture>> leases = OOCFuture.allOf(futures, StoreLease::close); + OOCStream.QueueCallback pendingAnchor = anchor; + ReservationBudget pendingBudget = budget; + anchor = null; + budget = null; + active = false; + leases.whenComplete( + (values, error) -> onJoinReady(pendingAnchor, callbackIndex, values, pendingBudget, error)); + } + finally { + if(anchor != null) + anchor.close(); + if(budget != null) + budget.close(); + if(active) + completeActive(); + } + } + + private void onJoinReady(OOCStream.QueueCallback anchor, int anchorIndex, + List> leases, ReservationBudget budget, Throwable error) { + JoinWork work = null; + try { + if(error != null) + throw DMLRuntimeException.of(error); + work = new JoinWork(anchor, anchorIndex, leases, budget); + _ready.enqueue(work); + work = null; + } + catch(Throwable failure) { + fail(failure); + } + finally { + if(work != null) + work.close(); + if(error != null) { + anchor.close(); + budget.close(); + } + completeActive(); + } + } + + private void process(JoinWork work) { + List values = new ArrayList<>(work._leases.size() + 1); + int lease = 0; + for(int i = 0; i <= work._leases.size(); i++) + values.add(i == work._anchorIndex ? work._anchor.get() : work._leases.get(lease++).value()); + IndexedMatrixValue output = _operation.apply(values); + long bytes = _size.applyAsLong(output); + work._budget.reserveBlocking(bytes); + OOCStream.QueueCallback callback = new InMemoryQueueCallback<>(output, null, work._budget, + bytes); + try { + _outputStream.enqueue(callback); + callback = null; + } + finally { + if(callback != null) + callback.close(); + } + } + + private void completeActive() { + if(_active.decrementAndGet() != 0) + return; + try { + _table.close(); + try { + _ready.closeInput(); + } + catch(IllegalStateException ignored) { + } + } + finally { + _activeCompletion.complete(null); + } + } + + private static final class JoinWork implements AutoCloseable { + private final OOCStream.QueueCallback _anchor; + private final int _anchorIndex; + private final List> _leases; + private final ReservationBudget _budget; + + private JoinWork(OOCStream.QueueCallback anchor, int anchorIndex, + List> leases, ReservationBudget budget) { + _anchor = anchor; + _anchorIndex = anchorIndex; + _leases = leases; + _budget = budget; + } + + @Override + public void close() { + _anchor.close(); + for(StoreLease lease : _leases) + lease.close(); + _budget.close(); + } + } +} diff --git a/src/main/java/org/apache/sysds/runtime/ooc/store/StateTable.java b/src/main/java/org/apache/sysds/runtime/ooc/store/StateTable.java index b70a7f4988a..2abdd329667 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/store/StateTable.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/store/StateTable.java @@ -19,9 +19,11 @@ package org.apache.sysds.runtime.ooc.store; +import org.apache.sysds.runtime.instructions.ooc.CachingStream; import org.apache.sysds.runtime.ooc.cache.BlockEntry; import org.apache.sysds.runtime.ooc.cache.BlockKey; import org.apache.sysds.runtime.ooc.cache.OOCCache; +import org.apache.sysds.runtime.ooc.cache.OOCCacheManager; import org.apache.sysds.runtime.ooc.cache.OOCFuture; import org.apache.sysds.runtime.ooc.cache.io.SpillableObject; import org.apache.sysds.runtime.ooc.memory.ManagedPayload; @@ -49,6 +51,14 @@ public final class StateTable implements AutoCloseabl private volatile AtomicIntegerArray _generationSlots; private volatile boolean _closed; + public StateTable() { + this(OOCCacheManager.getGlobalCache(), CachingStream._streamSeq.getNextID()); + } + + public StateTable(int numSlots) { + this(OOCCacheManager.getGlobalCache(), CachingStream._streamSeq.getNextID(), numSlots); + } + public StateTable(OOCCache cache, long streamId) { this(cache, streamId, INITIAL_SLOTS); } diff --git a/src/main/java/org/apache/sysds/runtime/ooc/util/OOCInstructionUtils.java b/src/main/java/org/apache/sysds/runtime/ooc/util/OOCInstructionUtils.java index 3eedb32b8e6..2d5fa50d787 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/util/OOCInstructionUtils.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/util/OOCInstructionUtils.java @@ -40,13 +40,16 @@ import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue; import org.apache.sysds.runtime.matrix.data.MatrixBlock; import org.apache.sysds.runtime.matrix.data.MatrixIndexes; +import org.apache.sysds.runtime.ooc.cache.OOCCacheManager; import org.apache.sysds.runtime.ooc.cache.OOCFuture; +import org.apache.sysds.runtime.ooc.cache.io.SpillableObject; import org.apache.sysds.runtime.ooc.memory.MemoryAllowance; import org.apache.sysds.runtime.ooc.memory.ReservationBudget; import org.apache.sysds.runtime.ooc.primitives.BroadcastOOCPrimitive; import org.apache.sysds.runtime.ooc.primitives.GroupedReduceOOCPrimitive; import org.apache.sysds.runtime.ooc.primitives.JoinOOCPrimitive; import org.apache.sysds.runtime.ooc.primitives.MappingOOCPrimitive; +import org.apache.sysds.runtime.ooc.primitives.NaryJoinOOCPrimitive; import org.apache.sysds.runtime.ooc.primitives.PlannableDataGenOOCPrimitive; import org.apache.sysds.runtime.ooc.primitives.ReduceOOCPrimitive; import org.apache.sysds.runtime.ooc.primitives.TransposeOOCPrimitive; @@ -92,7 +95,39 @@ public static void transpose(OOCStreamable input, OOCStream< public static void equiJoin(OOCStreamable left, OOCStreamable right, OOCStream output, BiFunction operation, StreamContext context) { - output.assignPrimitive(new JoinOOCPrimitive(left, right, output, operation, context)); + long cols = right.getDataCharacteristics().getNumColBlocks(); + long inputBytes = Math.max(OOCUtils.estimateOutputTileBytes(left.getDataCharacteristics()), + OOCUtils.estimateOutputTileBytes(right.getDataCharacteristics())); + long outputBytes = OOCUtils.estimateOutputTileBytes(output.getDataCharacteristics()); + ToIntFunction key = value -> Math + .toIntExact((value.getIndexes().getRowIndex() - 1) * cols + value.getIndexes().getColumnIndex() - 1); + keyedJoin(left, right, output, key, key, value -> ((MatrixBlock) value.getValue()).getExactSerializedSize(), + (leftValue, rightValue) -> new IndexedMatrixValue(leftValue.getIndexes(), + operation.apply((MatrixBlock) leftValue.getValue(), (MatrixBlock) rightValue.getValue())), + inputBytes + OOCCacheManager.getGlobalCache().maxPhysicalPinBytes(inputBytes) + outputBytes, context); + } + + public static void keyedJoin(OOCStreamable left, + OOCStreamable right, OOCStream output, ToIntFunction leftKey, ToIntFunction rightKey, + ToLongFunction outputSize, BiFunction operation, long taskBytes, StreamContext context) { + output.assignPrimitive( + new JoinOOCPrimitive<>(left, right, output, leftKey, rightKey, outputSize, operation, taskBytes, context)); + } + + public static void naryEquiJoin(List> inputs, + OOCStream output, Function, IndexedMatrixValue> operation, + StreamContext context) { + long cols = inputs.get(0).getDataCharacteristics().getNumColBlocks(); + long inputBytes = inputs.stream().map(OOCStreamable::getDataCharacteristics) + .mapToLong(OOCUtils::estimateOutputTileBytes).max().orElse(0); + long outputBytes = OOCUtils.estimateOutputTileBytes(output.getDataCharacteristics()); + ToIntFunction key = value -> Math + .toIntExact((value.getIndexes().getRowIndex() - 1) * cols + value.getIndexes().getColumnIndex() - 1); + ToLongFunction size = value -> ((MatrixBlock) value.getValue()).getExactSerializedSize(); + long joinBytes = (inputs.size() - 1) * OOCCacheManager.getGlobalCache().maxPhysicalPinBytes(inputBytes) + + outputBytes; + output.assignPrimitive( + new NaryJoinOOCPrimitive(inputs, output, key, size, operation, inputBytes, joinBytes, context)); } public static void indexedBroadcastMap(OOCStreamable streamed, diff --git a/src/main/java/org/apache/sysds/runtime/ooc/util/StateTableUtils.java b/src/main/java/org/apache/sysds/runtime/ooc/util/StateTableUtils.java index 641192c16a6..97978b7eaaa 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/util/StateTableUtils.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/util/StateTableUtils.java @@ -20,9 +20,8 @@ package org.apache.sysds.runtime.ooc.util; import org.apache.sysds.runtime.instructions.ooc.OOCStream; -import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue; -import org.apache.sysds.runtime.matrix.data.MatrixBlock; import org.apache.sysds.runtime.ooc.cache.OOCFuture; +import org.apache.sysds.runtime.ooc.cache.io.SpillableObject; import org.apache.sysds.runtime.ooc.memory.InMemoryQueueCallback; import org.apache.sysds.runtime.ooc.memory.ManagedPayload; import org.apache.sysds.runtime.ooc.memory.MemoryAllowance; @@ -31,24 +30,64 @@ import org.apache.sysds.runtime.ooc.store.StoreLease; public final class StateTableUtils { - public static OOCFuture putOrTake(StateTable table, int slot, - OOCStream.QueueCallback tile, MemoryAllowance allowance) { - if(tile instanceof MaterializedCallback pinned && pinned.pinnedEntry() != null) - return putReferenceOrTake(table, slot, pinned, allowance); - ManagedPayload payload; - if(tile instanceof InMemoryQueueCallback managed && managed.getManagedBytes() > 0) { + public static OOCFuture> take(StateTable table, int slot, + MemoryAllowance allowance) { + OOCFuture> future = table.take(slot, allowance); + OOCFuture> toReturn = new OOCFuture<>(); + future.whenComplete((l, err) -> { + if(err != null) + toReturn.completeExceptionally(err); + else + toReturn.complete(new MaterializedCallback<>(l)); + }); + return toReturn; + } + + public static void put(StateTable table, int slot, OOCStream.QueueCallback tile, + MemoryAllowance allowance) { + if(tile instanceof MaterializedCallback pinned && pinned.pinnedEntry() != null) { + table.putReference(slot, pinned.pinnedEntry()); + return; + } + ManagedPayload payload; + if(tile instanceof InMemoryQueueCallback managed && managed.getManagedBytes() > 0) + payload = managed.extractManagedPayload(); + else { + T value = tile.get(); + long bytes = value.size(); + allowance.reserveBlocking(bytes); + payload = new ManagedPayload<>(value, bytes, allowance); + } + try { + table.put(slot, payload); + } + catch(RuntimeException error) { + payload.release(); + throw error; + } + } + + public static OOCFuture> putOrTake(StateTable table, int slot, + OOCStream.QueueCallback tile, MemoryAllowance allowance) { + if(tile instanceof MaterializedCallback pinned && pinned.pinnedEntry() != null) { + MaterializedCallback retained = (MaterializedCallback) pinned.keepOpen(); + pinned.close(); + return putReferenceOrTake(table, slot, retained, allowance); + } + ManagedPayload payload; + if(tile instanceof InMemoryQueueCallback managed && managed.getManagedBytes() > 0) { payload = managed.extractManagedPayload(); managed.close(); } else { - IndexedMatrixValue value = tile.get(); - long bytes = ((MatrixBlock) value.getValue()).getExactSerializedSize(); + T value = tile.get(); + long bytes = value.size(); allowance.reserveBlocking(bytes); payload = new ManagedPayload<>(value, bytes, allowance); tile.close(); } - OOCFuture result = new OOCFuture<>(); - OOCFuture> matched; + OOCFuture> result = new OOCFuture<>(); + OOCFuture> matched; try { matched = table.putOrTake(slot, payload, allowance); } @@ -65,16 +104,16 @@ else if(lease == null) result.complete(null); else result.complete( - new Match(new MaterializedCallback<>(StoreLease.create(payload.value(), payload::release)), + new Match<>(new MaterializedCallback<>(StoreLease.create(payload.value(), payload::release)), new MaterializedCallback<>(lease))); }); return result; } - private static OOCFuture putReferenceOrTake(StateTable table, int slot, - MaterializedCallback pinned, MemoryAllowance allowance) { - OOCFuture result = new OOCFuture<>(); - OOCFuture> matched; + private static OOCFuture> putReferenceOrTake(StateTable table, int slot, + MaterializedCallback pinned, MemoryAllowance allowance) { + OOCFuture> result = new OOCFuture<>(); + OOCFuture> matched; try { matched = table.putReferenceOrTake(slot, pinned.pinnedEntry(), allowance); } @@ -92,12 +131,11 @@ else if(lease == null) { result.complete(null); } else - result.complete(new Match(pinned, new MaterializedCallback<>(lease))); + result.complete(new Match<>(pinned, new MaterializedCallback<>(lease))); }); return result; } - public record Match(OOCStream.QueueCallback left, - OOCStream.QueueCallback right) { + public record Match(OOCStream.QueueCallback left, OOCStream.QueueCallback right) { } } diff --git a/src/test/java/org/apache/sysds/test/component/ooc/OOCPrimitiveTest.java b/src/test/java/org/apache/sysds/test/component/ooc/OOCPrimitiveTest.java index 336a7aecfd0..b408d3f1e9e 100644 --- a/src/test/java/org/apache/sysds/test/component/ooc/OOCPrimitiveTest.java +++ b/src/test/java/org/apache/sysds/test/component/ooc/OOCPrimitiveTest.java @@ -249,6 +249,43 @@ private static Map runGroupedReduce(GroupedReduceOOCPrimitive.Gr return values; } + @Test + public void testNaryJoinOutOfOrder() { + SubscribableTaskQueue first = new SubscribableTaskQueue<>(); + SubscribableTaskQueue second = new SubscribableTaskQueue<>(); + SubscribableTaskQueue third = new SubscribableTaskQueue<>(); + SubscribableTaskQueue output = new SubscribableTaskQueue<>(); + for(SubscribableTaskQueue stream : List.of(first, second, third, output)) + stream.setData(new MatrixObject(ValueType.FP64, "/dev/null", + new MetaDataFormat(new MatrixCharacteristics(1, 2, 1), FileFormat.BINARY))); + CachingStream cachedSecond = new CachingStream(second); + first.enqueue(new IndexedMatrixValue(new MatrixIndexes(1, 1), new MatrixBlock(1, 1, 10d))); + first.enqueue(new IndexedMatrixValue(new MatrixIndexes(1, 2), new MatrixBlock(1, 1, 20d))); + second.enqueue(new IndexedMatrixValue(new MatrixIndexes(1, 2), new MatrixBlock(1, 1, 2d))); + second.enqueue(new IndexedMatrixValue(new MatrixIndexes(1, 1), new MatrixBlock(1, 1, 1d))); + third.enqueue(new IndexedMatrixValue(new MatrixIndexes(1, 1), new MatrixBlock(1, 1, 100d))); + third.enqueue(new IndexedMatrixValue(new MatrixIndexes(1, 2), new MatrixBlock(1, 1, 200d))); + first.closeInput(); + second.closeInput(); + third.closeInput(); + + OOCInstructionUtils.naryEquiJoin(List.of(first, cachedSecond, third), output, + blocks -> new IndexedMatrixValue(blocks.get(0).getIndexes(), + new MatrixBlock(1, 1, blocks.get(0).getValue().get(0, 0) + 10 * blocks.get(1).getValue().get(0, 0) + + 100 * blocks.get(2).getValue().get(0, 0))), + new StreamContext()); + + output.start(); + Map values = new HashMap<>(); + OOCStream.QueueCallback callback; + while((callback = output.dequeueCB()) != null) + try(OOCStream.QueueCallback current = callback) { + values.put(current.get().getIndexes().getColumnIndex(), current.get().getValue().get(0, 0)); + } + Assert.assertEquals(Map.of(1L, 10020d, 2L, 20040d), values); + cachedSecond.scheduleDeletion(); + } + @Test public void testJoinOutOfOrder() { SubscribableTaskQueue left = new SubscribableTaskQueue<>(); diff --git a/src/test/java/org/apache/sysds/test/functions/ooc/TernaryMatrixTest.java b/src/test/java/org/apache/sysds/test/functions/ooc/TernaryMatrixTest.java new file mode 100644 index 00000000000..55791ec0160 --- /dev/null +++ b/src/test/java/org/apache/sysds/test/functions/ooc/TernaryMatrixTest.java @@ -0,0 +1,105 @@ +/* + * 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.sysds.test.functions.ooc; + +import java.io.IOException; + +import org.apache.sysds.common.Opcodes; +import org.apache.sysds.common.Types; +import org.apache.sysds.runtime.instructions.Instruction; +import org.apache.sysds.runtime.io.MatrixWriter; +import org.apache.sysds.runtime.io.MatrixWriterFactory; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.meta.MatrixCharacteristics; +import org.apache.sysds.runtime.util.DataConverter; +import org.apache.sysds.runtime.util.HDFSTool; +import org.apache.sysds.test.AutomatedTestBase; +import org.apache.sysds.test.TestConfiguration; +import org.apache.sysds.test.TestUtils; +import org.junit.Assert; +import org.junit.Test; + +public class TernaryMatrixTest extends AutomatedTestBase { + private static final String TEST_NAME = "TernaryMatrix"; + private static final String TEST_DIR = "functions/ooc/"; + private static final String TEST_CLASS_DIR = TEST_DIR + TernaryMatrixTest.class.getSimpleName() + "/"; + private static final int ROWS = 1200; + private static final int COLS = 1100; + private static final int BLOCK_SIZE = 1000; + + @Override + public void setUp() { + TestUtils.clearAssertionInformation(); + addTestConfiguration(TEST_NAME, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME)); + } + + @Test + public void testTernaryOperations() throws IOException { + Types.ExecMode oldPlatform = setExecMode(Types.ExecMode.SINGLE_NODE); + try { + getAndLoadTestConfiguration(TEST_NAME); + fullDMLScriptName = SCRIPT_DIR + TEST_DIR + TEST_NAME + ".dml"; + writeInput("A", MatrixBlock.randOperations(ROWS, COLS, 1, -1, 1, "uniform", 7)); + writeInput("B", MatrixBlock.randOperations(ROWS, COLS, 0.7, -2, 2, "uniform", 8)); + writeInput("C", MatrixBlock.randOperations(ROWS, COLS, 0.2, -3, 3, "uniform", 9)); + + String[] outputs = {"plus", "minus", "ifelse"}; + Opcodes[] opcodes = {Opcodes.PM, Opcodes.MINUSMULT, Opcodes.IFELSE}; + for(int i = 0; i < outputs.length; i++) { + programArgs = arguments(true, i + 1, outputs[i]); + runTest(true, false, null, -1); + Assert.assertTrue(heavyHittersContainsString(Instruction.OOC_INST_PREFIX + opcodes[i])); + + programArgs = arguments(false, i + 1, outputs[i] + "_target"); + runTest(true, false, null, -1); + MatrixBlock actual = DataConverter.readMatrixFromHDFS(output(outputs[i]), Types.FileFormat.BINARY, ROWS, + COLS, BLOCK_SIZE); + MatrixBlock expected = DataConverter.readMatrixFromHDFS(output(outputs[i] + "_target"), + Types.FileFormat.BINARY, ROWS, COLS, BLOCK_SIZE); + TestUtils.compareMatrices(actual, expected, 1e-8); + } + } + finally { + resetExecMode(oldPlatform); + } + } + + private String[] arguments(boolean ooc, int operation, String result) { + String[] args = new String[ooc ? 8 : 7]; + int offset = 0; + args[offset++] = "-stats"; + if(ooc) + args[offset++] = "-ooc"; + args[offset++] = "-args"; + args[offset++] = input("A"); + args[offset++] = input("B"); + args[offset++] = input("C"); + args[offset++] = Integer.toString(operation); + args[offset] = output(result); + return args; + } + + private void writeInput(String name, MatrixBlock value) throws IOException { + MatrixWriter writer = MatrixWriterFactory.createMatrixWriter(Types.FileFormat.BINARY); + writer.writeMatrixToHDFS(value, input(name), ROWS, COLS, BLOCK_SIZE, value.getNonZeros()); + HDFSTool.writeMetaDataFile(input(name + ".mtd"), Types.ValueType.FP64, + new MatrixCharacteristics(ROWS, COLS, BLOCK_SIZE, value.getNonZeros()), Types.FileFormat.BINARY); + } +} diff --git a/src/test/scripts/functions/ooc/TernaryMatrix.dml b/src/test/scripts/functions/ooc/TernaryMatrix.dml new file mode 100644 index 00000000000..aef0b901973 --- /dev/null +++ b/src/test/scripts/functions/ooc/TernaryMatrix.dml @@ -0,0 +1,36 @@ +#------------------------------------------------------------- +# +# 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. +# +#------------------------------------------------------------- + +A = read($1); +B = read($2); +C = read($3); + +if($4 == 1) { + result = A + 2 * B; +} +else if($4 == 2) { + result = A - 2 * B; +} +else { + result = ifelse(A > 0, B, C); +} + +write(result, $5, format="binary");