Java 类org.apache.commons.lang3.mutable.MutableObject 实例源码

/**
 * Result {@code null} means that there is no variable. Result other than {@code null} means that there is a variable (that may possibly
 * be {@code null}).
 *
 * @param name the name of the variable.
 * @return a holder for a variable.
 */
protected Mutable<Object> doGetVariable(String name) {
    List<Object> variables =
            scripts.stream().filter(script -> script.getMetaClass().hasProperty(script.getMetaClass().getTheClass(), name) != null)
                    .map(script -> script.getProperty(name)).collect(Collectors.toList());

    if (variables.isEmpty()) {
        try {
            return new MutableObject<>(binding.getProperty(name));
        } catch (MissingPropertyException e) {
            return null; // This means that no variable has been found!
        }
    }

    return new MutableObject<>(variables.get(0));
}

@Test
public void shouldAllowSameMethodRegistrationFromDifferentThreads() throws Exception {

    final MutableObject<StateMachineDefinition> definitionOne = new MutableObject<>(null);
    final MutableObject<StateMachineDefinition> definitionTwo = new MutableObject<>(null);

    final Thread thread1 = new Thread(() -> {
        localContext.registerNew("fooBar", 1, "someDescription","someContext");
        definitionOne.setValue(tlStateMachineDef.get());
    });
    final Thread thread2 = new Thread(() -> {
        localContext.registerNew("fooBar", 1, "someDescription","someContext");
        definitionTwo.setValue(tlStateMachineDef.get());
    });
    thread1.start();
    thread2.start();

    thread1.join();
    thread2.join();

    assertThat(definitionOne.getValue()).isNotNull().isEqualTo(definitionTwo.getValue()).isEqualTo(new StateMachineDefinition("someDescription", "fooBar", 1l, new HashSet<>(), new HashSet<>(), "someContext"));
}

@JtwigFunction(name = "render")
public <T extends RenderedElement> String render(HttpServletRequest request, @Parameter T input) throws FunctionException {
    try {
        if (input == null || !input.isVisible()) {
            return "";
        }
        RenderedElementService renderedElementService = applicationContext.getBean(RenderedElementService.class);
        HookService hookService = applicationContext.getBean(HookService.class);
        final MutableObject<RenderedElement> elementToRender = new MutableObject<>(input);
        if (!input.isWrapped()) {
            hookService.executeHook("element_wrapper", (c, o) -> {
                if (o != null && o instanceof RenderedElement && !input.isWrapped()) {
                    input.setWrapped(true);
                    elementToRender.setValue((RenderedElement) o);
                }
            }, input);
        }
        hookService.executeHook("element_alter", elementToRender.getValue());
        ModelAndView modelAndView = renderedElementService.convertToModelAndView(elementToRender.getValue());
        return render(request, modelAndView);
    } catch (Exception ex) {
        throw new FunctionException(ex);
    }
}

private AggregateOperator getAggregateOperator(IFunctionInfo aggregateFunction,
        Mutable<ILogicalExpression> aggregateArgs, LogicalVariable aggregateVariable) {
    List<LogicalVariable> aggregateVariables = new ArrayList<LogicalVariable>();
    aggregateVariables.add(aggregateVariable);

    List<Mutable<ILogicalExpression>> aggregateSequenceArgs = new ArrayList<Mutable<ILogicalExpression>>();
    aggregateSequenceArgs.add(aggregateArgs);

    List<Mutable<ILogicalExpression>> exprs = new ArrayList<Mutable<ILogicalExpression>>();
    ILogicalExpression aggregateExp = new AggregateFunctionCallExpression(aggregateFunction, false,
            aggregateSequenceArgs);
    Mutable<ILogicalExpression> aggregateExpRef = new MutableObject<ILogicalExpression>(aggregateExp);
    exprs.add(aggregateExpRef);

    return new AggregateOperator(aggregateVariables, exprs);
}

private void rewriteGroupByAggregate(LogicalVariable oldAggVar, GroupByOperator gbyOp,
        AggregateFunctionCallExpression aggFun, LogicalVariable newAggVar, IOptimizationContext context)
                throws AlgebricksException {
    for (int j = 0; j < gbyOp.getNestedPlans().size(); j++) {
        AggregateOperator aggOp = (AggregateOperator) gbyOp.getNestedPlans().get(j).getRoots().get(0).getValue();
        int n = aggOp.getVariables().size();
        for (int i = 0; i < n; i++) {
            LogicalVariable v = aggOp.getVariables().get(i);
            if (v.equals(oldAggVar)) {
                AbstractFunctionCallExpression oldAggExpr = (AbstractFunctionCallExpression) aggOp.getExpressions()
                        .get(i).getValue();
                AggregateFunctionCallExpression newAggFun = new AggregateFunctionCallExpression(
                        aggFun.getFunctionInfo(), false, new ArrayList<Mutable<ILogicalExpression>>());
                for (Mutable<ILogicalExpression> arg : oldAggExpr.getArguments()) {
                    ILogicalExpression cloned = ((AbstractLogicalExpression) arg.getValue()).cloneExpression();
                    newAggFun.getArguments().add(new MutableObject<ILogicalExpression>(cloned));
                }
                aggOp.getVariables().add(newAggVar);
                aggOp.getExpressions().add(new MutableObject<ILogicalExpression>(newAggFun));
                context.computeAndSetTypeEnvironmentForOperator(aggOp);
                break;
            }
        }
    }
}

private AggregateOperator getAggregateOperator(LogicalVariable unnestVariable, LogicalVariable aggregateVariable) {
    List<LogicalVariable> aggregateVariables = new ArrayList<LogicalVariable>();
    aggregateVariables.add(aggregateVariable);

    List<Mutable<ILogicalExpression>> aggregateSequenceArgs = new ArrayList<Mutable<ILogicalExpression>>();
    Mutable<ILogicalExpression> unnestVariableRef = new MutableObject<ILogicalExpression>(
            new VariableReferenceExpression(unnestVariable));
    aggregateSequenceArgs.add(unnestVariableRef);

    List<Mutable<ILogicalExpression>> exprs = new ArrayList<Mutable<ILogicalExpression>>();
    ILogicalExpression aggregateExp = new AggregateFunctionCallExpression(BuiltinOperators.SEQUENCE, false,
            aggregateSequenceArgs);
    Mutable<ILogicalExpression> aggregateExpRef = new MutableObject<ILogicalExpression>(aggregateExp);
    exprs.add(aggregateExpRef);

    return new AggregateOperator(aggregateVariables, exprs);
}

public static Result traverse( SchemaAST root, String path ) {
    SchemaAST schemaAST = root;
    val additionalProperties = new MutableObject<Boolean>( null );

    final Supplier<Result> empty = () -> new Result( Optional.empty(), Optional.ofNullable( additionalProperties.getValue() ) );

    for( val item : StringUtils.split( path, '.' ) ) {
        if( schemaAST instanceof ObjectSchemaAST ) {
            val objectSchemaAST = ( ObjectSchemaAST ) schemaAST;
            schemaAST = objectSchemaAST.properties.get( item );
            objectSchemaAST.additionalProperties.ifPresent( additionalProperties::setValue );
            if( schemaAST == null )
                return empty.get();
        } else if( schemaAST instanceof ArraySchemaAST ) {
            if( !"items".equals( item ) )
                return empty.get();
            schemaAST = ( ( ArraySchemaAST ) schemaAST ).items;
        } else {
            return empty.get();
        }
    }

    return new Result( Optional.of( schemaAST ), Optional.ofNullable( additionalProperties.getValue() ) );
}

private Map<LogicalVariable, LogicalVariable> buildVarExprList(Collection<LogicalVariable> vars,
        IOptimizationContext context, GroupByOperator g,
        List<Pair<LogicalVariable, Mutable<ILogicalExpression>>> outVeList) throws AlgebricksException {
    Map<LogicalVariable, LogicalVariable> m = new HashMap<LogicalVariable, LogicalVariable>();
    for (LogicalVariable ov : vars) {
        LogicalVariable newVar = context.newVar();
        ILogicalExpression varExpr = new VariableReferenceExpression(newVar);
        outVeList.add(new Pair<LogicalVariable, Mutable<ILogicalExpression>>(ov,
                new MutableObject<ILogicalExpression>(varExpr)));
        for (ILogicalPlan p : g.getNestedPlans()) {
            for (Mutable<ILogicalOperator> r : p.getRoots()) {
                OperatorManipulationUtil.substituteVarRec((AbstractLogicalOperator) r.getValue(), ov, newVar, true,
                        context);
            }
        }
        AbstractLogicalOperator opUnder = (AbstractLogicalOperator) g.getInputs().get(0).getValue();
        OperatorManipulationUtil.substituteVarRec(opUnder, ov, newVar, true, context);
        m.put(ov, newVar);
    }
    return m;
}

private AssignOperator createAssignBelowUnionAllBranch(UnionAllOperator unionOp, int inputIndex,
        AssignOperator originalAssignOp, Set<LogicalVariable> assignUsedVars, IOptimizationContext context)
                throws AlgebricksException {
    AssignOperator newAssignOp = cloneAssignOperator(originalAssignOp, context);
    newAssignOp.getInputs()
            .add(new MutableObject<ILogicalOperator>(unionOp.getInputs().get(inputIndex).getValue()));
    unionOp.getInputs().get(inputIndex).setValue(newAssignOp);
    int numVarMappings = unionOp.getVariableMappings().size();
    for (int i = 0; i < numVarMappings; i++) {
        Triple<LogicalVariable, LogicalVariable, LogicalVariable> varMapping = unionOp.getVariableMappings().get(i);
        if (assignUsedVars.contains(varMapping.third)) {
            LogicalVariable replacementVar;
            if (inputIndex == 0) {
                replacementVar = varMapping.first;
            } else {
                replacementVar = varMapping.second;
            }
            VariableUtilities.substituteVariables(newAssignOp, varMapping.third, replacementVar, context);
        }
    }
    context.computeAndSetTypeEnvironmentForOperator(newAssignOp);
    return newAssignOp;
}

private ILogicalOperator buildOperatorChain(List<ILogicalOperator> ops, ILogicalOperator bottomOp,
        IOptimizationContext context) throws AlgebricksException {
    ILogicalOperator root = ops.get(0);
    ILogicalOperator prevOp = root;
    for (int i = 1; i < ops.size(); i++) {
        ILogicalOperator inputOp = ops.get(i);
        prevOp.getInputs().clear();
        prevOp.getInputs().add(new MutableObject<ILogicalOperator>(inputOp));
        prevOp = inputOp;
    }
    if (bottomOp != null) {
        context.computeAndSetTypeEnvironmentForOperator(bottomOp);
        prevOp.getInputs().clear();
        prevOp.getInputs().add(new MutableObject<ILogicalOperator>(bottomOp));
    }
    return root;
}

private Mutable<ILogicalOperator> enforceOrderProperties(List<LocalOrderProperty> oList,
        Mutable<ILogicalOperator> topOp, boolean isMicroOp, IOptimizationContext context)
                throws AlgebricksException {
    List<Pair<IOrder, Mutable<ILogicalExpression>>> oe = new LinkedList<Pair<IOrder, Mutable<ILogicalExpression>>>();
    for (LocalOrderProperty orderProperty : oList) {
        for (OrderColumn oc : orderProperty.getOrderColumns()) {
            IOrder ordType = (oc.getOrder() == OrderKind.ASC) ? OrderOperator.ASC_ORDER : OrderOperator.DESC_ORDER;
            Pair<IOrder, Mutable<ILogicalExpression>> pair = new Pair<IOrder, Mutable<ILogicalExpression>>(ordType,
                    new MutableObject<ILogicalExpression>(new VariableReferenceExpression(oc.getColumn())));
            oe.add(pair);
        }
    }
    OrderOperator oo = new OrderOperator(oe);
    oo.setExecutionMode(AbstractLogicalOperator.ExecutionMode.LOCAL);
    if (isMicroOp) {
        oo.setPhysicalOperator(new InMemoryStableSortPOperator());
    } else {
        oo.setPhysicalOperator(new StableSortPOperator(physicalOptimizationConfig.getMaxFramesExternalSort()));
    }
    oo.getInputs().add(topOp);
    context.computeAndSetTypeEnvironmentForOperator(oo);
    if (AlgebricksConfig.DEBUG) {
        AlgebricksConfig.ALGEBRICKS_LOGGER.fine(">>>> Added sort enforcer " + oo.getPhysicalOperator() + ".\n");
    }
    return new MutableObject<ILogicalOperator>(oo);
}

private static boolean pushAllProjectionsOnTopOf(Collection<LogicalVariable> toPush,
        Mutable<ILogicalOperator> opRef, IOptimizationContext context, ILogicalOperator initialOp)
                throws AlgebricksException {
    if (toPush.isEmpty()) {
        return false;
    }
    AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();

    if (context.checkAndAddToAlreadyCompared(initialOp, op)) {
        return false;
    }

    if (op.getOperatorTag() == LogicalOperatorTag.PROJECT) {
        return false;
    }

    ProjectOperator pi2 = new ProjectOperator(new ArrayList<LogicalVariable>(toPush));
    pi2.getInputs().add(new MutableObject<ILogicalOperator>(op));
    opRef.setValue(pi2);
    pi2.setExecutionMode(op.getExecutionMode());
    context.computeAndSetTypeEnvironmentForOperator(pi2);
    return true;
}

@Override
public ILogicalOperator visitIndexInsertDeleteUpsertOperator(IndexInsertDeleteUpsertOperator op, Void arg)
        throws AlgebricksException {
    List<Mutable<ILogicalExpression>> newPrimaryKeyExpressions = new ArrayList<Mutable<ILogicalExpression>>();
    deepCopyExpressionRefs(newPrimaryKeyExpressions, op.getPrimaryKeyExpressions());
    List<Mutable<ILogicalExpression>> newSecondaryKeyExpressions = new ArrayList<Mutable<ILogicalExpression>>();
    deepCopyExpressionRefs(newSecondaryKeyExpressions, op.getSecondaryKeyExpressions());
    Mutable<ILogicalExpression> newFilterExpression = new MutableObject<ILogicalExpression>(
            ((AbstractLogicalExpression) op.getFilterExpression()).cloneExpression());
    List<Mutable<ILogicalExpression>> newLSMComponentFilterExpressions = new ArrayList<Mutable<ILogicalExpression>>();
    deepCopyExpressionRefs(newLSMComponentFilterExpressions, op.getAdditionalFilteringExpressions());
    IndexInsertDeleteUpsertOperator indexInsertDeleteOp = new IndexInsertDeleteUpsertOperator(
            op.getDataSourceIndex(), newPrimaryKeyExpressions, newSecondaryKeyExpressions, newFilterExpression,
            op.getOperation(), op.isBulkload());
    indexInsertDeleteOp.setAdditionalFilteringExpressions(newLSMComponentFilterExpressions);
    return indexInsertDeleteOp;
}

@Override
public ILogicalOperator visitTokenizeOperator(TokenizeOperator op, Void arg) throws AlgebricksException {
    List<Mutable<ILogicalExpression>> newPrimaryKeyExpressions = new ArrayList<Mutable<ILogicalExpression>>();
    deepCopyExpressionRefs(newPrimaryKeyExpressions, op.getPrimaryKeyExpressions());
    List<Mutable<ILogicalExpression>> newSecondaryKeyExpressions = new ArrayList<Mutable<ILogicalExpression>>();
    deepCopyExpressionRefs(newSecondaryKeyExpressions, op.getSecondaryKeyExpressions());
    List<LogicalVariable> newTokenizeVars = new ArrayList<LogicalVariable>();
    deepCopyVars(newTokenizeVars, op.getTokenizeVars());
    Mutable<ILogicalExpression> newFilterExpression = new MutableObject<ILogicalExpression>(
            ((AbstractLogicalExpression) op.getFilterExpression()).cloneExpression());
    List<Object> newTokenizeVarTypes = new ArrayList<Object>();
    deepCopyObjects(newTokenizeVarTypes, op.getTokenizeVarTypes());

    TokenizeOperator tokenizeOp = new TokenizeOperator(op.getDataSourceIndex(), newPrimaryKeyExpressions,
            newSecondaryKeyExpressions, newTokenizeVars, newFilterExpression, op.getOperation(), op.isBulkload(),
            op.isPartitioned(), newTokenizeVarTypes);
    return tokenizeOp;
}

/**
 * Request a remote synchronization. This call blocks until the operation is complete,
 * or the thread is interrupted.
 * 
 * @throws InterruptedException
 */
public SynchronizeResult blockingSynchronize() throws InterruptedException
{
    log.info("Queuing a blocking sync request");
    final MutableObject<SynchronizeResult> result = new MutableObject<SynchronizeResult>();
    final CountDownLatch cdl = new CountDownLatch(1);
    Consumer<SynchronizeResult> callback = new Consumer<SynchronizeResult>()
    {
        @Override
        public void accept(SynchronizeResult t)
        {
            result.setValue(t);
            cdl.countDown();
        }
    };

    synchronize(callback);
    cdl.await();
    return result.getValue();
}

private Callable<Object> createServer(final MutableObject<String> actualRequest) {
    return new Callable<Object>() {
        @Override
        public Object call() throws Exception {
            LOGGER.info("SERVER: receiving request");
            UDPTransport transport = new UDPTransport(SERVER_PORT);
            String request = transport.receive();
            actualRequest.setValue(request);

            LOGGER.info("SERVER: sending answer");
            transport.send(ANSWER);

            LOGGER.info("SERVER: finished");
            return null;
        }
    };
}

private Callable<Object> createClient(final MutableObject<String> actualAnswer) {
    return new Callable<Object>() {
        @Override
        public Object call() throws Exception {
            LOGGER.info("CLIENT: sending request");
            UDPTransport transport = new UDPTransport(InetAddress.getLocalHost(), SERVER_PORT, TIMEOUT);
            transport.send(REQUEST);

            LOGGER.info("CLIENT: receiving answer");
            String answer = transport.receive();
            actualAnswer.setValue(answer);

            LOGGER.info("CLIENT: finished");
            return null;
        }
    };
}

@Override
public Object onEvent(T stateful, String event, Object... parms) throws TooBusyException {
    ArrayList<Object> parmList = new ArrayList<Object>(Arrays.asList(parms));

    // Create a Mutable Object and add it to the Parameter List - it will be used
    // to return the returned value from the Controller as the FSM returns the State
    //
    MutableObject<T> returnValue = new MutableObject<T>();
    ArrayList<Object> invokeParmlist = new ArrayList<Object>(parms.length + 1);
    invokeParmlist.add(returnValue);
    invokeParmlist.addAll(parmList);

    // Call the FSM
    // 
    fsm.onEvent(stateful, event, invokeParmlist.toArray());
    return returnValue.getValue();
}

private void evaluate(WebDriver driver, WebElementSelector locator, BiFunction<WebDriver, WebElementSelector, String> callback, boolean disableValueCheck, MutableObject<String> resultHolder) {
    String result = StringUtils.stripToEmpty(callback.apply(driver, locator));
    resultHolder.setValue(result);
    String expectedValue = locator.getExpectedValue();
    if (disableValueCheck || StringUtils.isBlank(expectedValue) || this.fitnesseMarkup.compare(expectedValue, result)) {
        return;
    }
    throw new NoSuchElementException(MessageFormat.format("Element with unexpected value [Expected: {0}, Obtained: {1}]", expectedValue, result));
}

@Test
public void testBeforeCommitOrClearTask() throws ServiceException, SecurityServiceException {
    TestEntity entity = new TestEntity("entity");
    testEntityService.create(entity);
    final Long entityId = entity.getId();

    final MutableObject<TestUseEntityBeforeCommitOrClearTask> taskReference = new MutableObject<>();

    writeTransactionTemplate.execute(new TransactionCallbackWithoutResult() {
        @Override
        protected void doInTransactionWithoutResult(TransactionStatus status) {
            TestEntity reloadedEntity = testEntityService.getById(entityId);

            // This task will fail if executed when the entity is not in the session anymore
            TestUseEntityBeforeCommitOrClearTask task = new TestUseEntityBeforeCommitOrClearTask(reloadedEntity);
            taskReference.setValue(task);

            transactionSynchronizationTaskManagerService.push(task);

            // Should trigger the task's execution
            transactionSynchronizationTaskManagerService.beforeClear();

            entityManagerClear();
        }
    });

    entityManagerClear();

    assertEquals(1, taskReference.getValue().getExecutionCount());
}

@Test
public void testBeforeCommitOrClearTaskWithRollback() throws ServiceException, SecurityServiceException {
    TestEntity entity = new TestEntity("entity");
    testEntityService.create(entity);
    final Long entityId = entity.getId();

    final MutableObject<TestUseEntityBeforeCommitOrClearTask> taskReference = new MutableObject<>();

    writeTransactionTemplate.execute(new TransactionCallbackWithoutResult() {
        @Override
        protected void doInTransactionWithoutResult(TransactionStatus status) {
            TestEntity reloadedEntity = testEntityService.getById(entityId);

            // This task will fail if executed when the entity is not in the session anymore
            TestUseEntityBeforeCommitOrClearTask task = new TestUseEntityBeforeCommitOrClearTask(reloadedEntity);
            taskReference.setValue(task);

            transactionSynchronizationTaskManagerService.push(task);

            // Should trigger the task's execution
            transactionSynchronizationTaskManagerService.beforeClear();

            entityManagerClear();

            status.setRollbackOnly();
        }
    });

    entityManagerClear();

    assertEquals(1, taskReference.getValue().getExecutionCount());
    assertEquals(1, taskReference.getValue().getRollbackCount());
}

public Order process(Order order) {
    log.info("Processing order {}", order);
    if (order.getMarketOrders() == null) {
        order.setMarketOrders(new ArrayList<>());
    }
    Mutable<BigDecimal> buyGrossAmount = new MutableObject<BigDecimal>();
    Mutable<BigDecimal> buyNetAmount = new MutableObject<BigDecimal>();
    createSellMarketOrders(order, buyGrossAmount, buyNetAmount);
    createBuyMarketOrders(order, buyGrossAmount.getValue(), buyNetAmount.getValue());
    return order;
}

private String getString(String name)
{
    int functionId = this.getNewId();
    this.sendString(String.format("GET %s %i\r\n\r\n", name, functionId));
    String pattern = String.format("RESP %i (.*)\r\n\r\n", functionId);
    MutableObject<String> result = new MutableObject<>(null);
    Consumer<String> callback = (input) ->
    {
        result.setValue(Pattern.compile(pattern).matcher(input).group(1));
    };
    this.awaitMessage("", callback);
    while(result.getValue() == null)
        this.getNextMessage();
    return result.getValue();
}

private UnnestOperator getUnnestOperator(LogicalVariable inputVariable, LogicalVariable unnestVariable) {
    VariableReferenceExpression inputVre = new VariableReferenceExpression(inputVariable);

    List<Mutable<ILogicalExpression>> iterateArgs = new ArrayList<Mutable<ILogicalExpression>>();
    iterateArgs.add(new MutableObject<ILogicalExpression>(inputVre));

    ILogicalExpression unnestExp = new UnnestingFunctionCallExpression(BuiltinOperators.ITERATE, iterateArgs);
    Mutable<ILogicalExpression> unnestExpRef = new MutableObject<ILogicalExpression>(unnestExp);

    return new UnnestOperator(unnestVariable, unnestExpRef);
}

@SuppressWarnings("unchecked")
private boolean convertAlgebricksExpression(Mutable<ILogicalExpression> searchM, IFunctionInfo funcInfo,
        boolean isBoolean) {
    AbstractFunctionCallExpression searchFunction = (AbstractFunctionCallExpression) searchM.getValue();
    searchFunction.setFunctionInfo(funcInfo);

    if (isBoolean) {
        ScalarFunctionCallExpression functionCallExp = new ScalarFunctionCallExpression(
                BuiltinFunctions.FN_BOOLEAN_1, new MutableObject<ILogicalExpression>(searchM.getValue()));
        searchM.setValue(functionCallExp);
    }
    return true;
}

private AssignOperator getAssignOperator(LogicalVariable unnestVariable, LogicalVariable outputVariable,
        IFunctionInfo inputFunction) {
    List<Mutable<ILogicalExpression>> nodeArgs = new ArrayList<Mutable<ILogicalExpression>>();
    nodeArgs.add(new MutableObject<ILogicalExpression>(new VariableReferenceExpression(unnestVariable)));
    ScalarFunctionCallExpression unctionExpression = new ScalarFunctionCallExpression(inputFunction, nodeArgs);
    Mutable<ILogicalExpression> nodeTreeIdExpression = new MutableObject<ILogicalExpression>(unctionExpression);
    return new AssignOperator(outputVariable, nodeTreeIdExpression);
}

private UnnestOperator getUnnestOperator(LogicalVariable inputVariable, LogicalVariable unnestVariable) {
    VariableReferenceExpression inputVre = new VariableReferenceExpression(inputVariable);

    List<Mutable<ILogicalExpression>> iterateArgs = new ArrayList<Mutable<ILogicalExpression>>();
    iterateArgs.add(new MutableObject<ILogicalExpression>(inputVre));

    ILogicalExpression unnestExp = new UnnestingFunctionCallExpression(BuiltinOperators.ITERATE, iterateArgs);
    Mutable<ILogicalExpression> unnestExpRef = new MutableObject<ILogicalExpression>(unnestExp);

    return new UnnestOperator(unnestVariable, unnestExpRef);
}

@Override
public JsonSerializer<?> findSerializer( SerializationConfig config, JavaType type, BeanDescription beanDesc ) {
    final Class<?> raw = type.getRawClass();
    if( MutableObject.class.isAssignableFrom( raw ) ) {
        return new MutableObjectSerializer( type );
    }

    return super.findSerializer( config, type, beanDesc );
}

@Override
@SuppressWarnings( "unchecked" )
public MutableObject deserialize( JsonParser p,
                                  DeserializationContext ctxt ) throws IOException {

    try {
        final MutableObject vc = ( MutableObject ) _valueClass.newInstance();

        vc.setValue( ctxt.readValue( p, refType ) );

        return vc;
    } catch( InstantiationException | IllegalAccessException e ) {
        throw ctxt.instantiationException( _valueClass, e );
    }
}

@Override
@SuppressWarnings( "unchecked" )
public MutableObject deserialize( JsonParser p, DeserializationContext ctxt, MutableObject intoValue ) throws IOException {
    intoValue.setValue( ctxt.readValue( p, refType ) );

    return intoValue;
}

@Override
public JsonDeserializer<?> findBeanDeserializer( JavaType type, DeserializationConfig config,
                                                 BeanDescription beanDesc ) throws JsonMappingException {
    final Class<?> raw = type.getRawClass();
    if( MutableObject.class.isAssignableFrom( raw ) ) {
        final Type[] actualTypeArguments = ( ( ParameterizedType ) raw.getGenericSuperclass() ).getActualTypeArguments();
        final JavaType refType = config.constructType( ( Class ) actualTypeArguments[0] );
        return new MutableObjectDeserializer( type, refType );
    }

    return super.findBeanDeserializer( type.getReferencedType(), config, beanDesc );
}

/**
 * Clones the given assign operator changing the returned variables to be new ones.
 * Also, leaves the inputs of the clone clear.
 */
private AssignOperator cloneAssignOperator(AssignOperator assignOp, IOptimizationContext context) {
    List<LogicalVariable> vars = new ArrayList<LogicalVariable>();
    List<Mutable<ILogicalExpression>> exprs = new ArrayList<Mutable<ILogicalExpression>>();
    int numVars = assignOp.getVariables().size();
    for (int i = 0; i < numVars; i++) {
        vars.add(context.newVar());
        exprs.add(new MutableObject<ILogicalExpression>(
                assignOp.getExpressions().get(i).getValue().cloneExpression()));
    }
    AssignOperator assignCloneOp = new AssignOperator(vars, exprs);
    assignCloneOp.setExecutionMode(assignOp.getExecutionMode());
    return assignCloneOp;
}

private final static void insertOneToOneExchange(Mutable<ILogicalOperator> i, IOptimizationContext context)
        throws AlgebricksException {
    ExchangeOperator e = new ExchangeOperator();
    e.setPhysicalOperator(new OneToOneExchangePOperator());
    ILogicalOperator inOp = i.getValue();

    e.getInputs().add(new MutableObject<ILogicalOperator>(inOp));
    i.setValue(e);
    // e.recomputeSchema();
    OperatorPropertiesUtil.computeSchemaAndPropertiesRecIfNull(e, context);
    ExecutionMode em = ((AbstractLogicalOperator) inOp).getExecutionMode();
    e.setExecutionMode(em);
    e.computeDeliveredPhysicalProperties(context);
    context.computeAndSetTypeEnvironmentForOperator(e);
}

private static boolean generateMergeAggregationExpressions(GroupByOperator gby, IOptimizationContext context)
        throws AlgebricksException {
    if (gby.getNestedPlans().size() != 1) {
        //External/Sort group-by currently works only for one nested plan with one root containing
        //an aggregate and a nested-tuple-source.
        throw new AlgebricksException(
                "External group-by currently works only for one nested plan with one root containing"
                        + "an aggregate and a nested-tuple-source.");
    }
    ILogicalPlan p0 = gby.getNestedPlans().get(0);
    if (p0.getRoots().size() != 1) {
        //External/Sort group-by currently works only for one nested plan with one root containing
        //an aggregate and a nested-tuple-source.
        throw new AlgebricksException(
                "External group-by currently works only for one nested plan with one root containing"
                        + "an aggregate and a nested-tuple-source.");
    }
    IMergeAggregationExpressionFactory mergeAggregationExpressionFactory = context
            .getMergeAggregationExpressionFactory();
    Mutable<ILogicalOperator> r0 = p0.getRoots().get(0);
    AbstractLogicalOperator r0Logical = (AbstractLogicalOperator) r0.getValue();
    if (r0Logical.getOperatorTag() != LogicalOperatorTag.AGGREGATE) {
        return false;
    }
    AggregateOperator aggOp = (AggregateOperator) r0.getValue();
    List<Mutable<ILogicalExpression>> aggFuncRefs = aggOp.getExpressions();
    List<LogicalVariable> originalAggVars = aggOp.getVariables();
    int n = aggOp.getExpressions().size();
    List<Mutable<ILogicalExpression>> mergeExpressionRefs = new ArrayList<Mutable<ILogicalExpression>>();
    for (int i = 0; i < n; i++) {
        ILogicalExpression mergeExpr = mergeAggregationExpressionFactory
                .createMergeAggregation(originalAggVars.get(i), aggFuncRefs.get(i).getValue(), context);
        if (mergeExpr == null) {
            return false;
        }
        mergeExpressionRefs.add(new MutableObject<ILogicalExpression>(mergeExpr));
    }
    aggOp.setMergeExpressions(mergeExpressionRefs);
    return true;
}

private boolean pushDownFunctions(AbstractBinaryJoinOperator joinOp, int inputIndex,
        List<Mutable<ILogicalExpression>> funcExprs, IOptimizationContext context) throws AlgebricksException {
    ILogicalOperator joinInputOp = joinOp.getInputs().get(inputIndex).getValue();
    liveVars.clear();
    VariableUtilities.getLiveVariables(joinInputOp, liveVars);
    Iterator<Mutable<ILogicalExpression>> funcIter = funcExprs.iterator();
    List<LogicalVariable> assignVars = null;
    List<Mutable<ILogicalExpression>> assignExprs = null;
    while (funcIter.hasNext()) {
        Mutable<ILogicalExpression> funcExprRef = funcIter.next();
        ILogicalExpression funcExpr = funcExprRef.getValue();
        usedVars.clear();
        funcExpr.getUsedVariables(usedVars);
        // Check if we can push the function down this branch.
        if (liveVars.containsAll(usedVars)) {
            if (assignVars == null) {
                assignVars = new ArrayList<LogicalVariable>();
                assignExprs = new ArrayList<Mutable<ILogicalExpression>>();
            }
            // Replace the original expression with a variable reference expression.
            LogicalVariable replacementVar = context.newVar();
            assignVars.add(replacementVar);
            assignExprs.add(new MutableObject<ILogicalExpression>(funcExpr));
            funcExprRef.setValue(new VariableReferenceExpression(replacementVar));
            funcIter.remove();
        }
    }
    // Create new assign operator below the join if any functions can be pushed.
    if (assignVars != null) {
        AssignOperator newAssign = new AssignOperator(assignVars, assignExprs);
        newAssign.getInputs().add(new MutableObject<ILogicalOperator>(joinInputOp));
        newAssign.setExecutionMode(joinOp.getExecutionMode());
        joinOp.getInputs().get(inputIndex).setValue(newAssign);
        context.computeAndSetTypeEnvironmentForOperator(newAssign);
        return true;
    }
    return false;
}

private void setNewOp(Mutable<ILogicalOperator> opRef, AbstractLogicalOperator newOp, IOptimizationContext context)
        throws AlgebricksException {
    ILogicalOperator oldOp = opRef.getValue();
    opRef.setValue(newOp);
    newOp.getInputs().add(new MutableObject<ILogicalOperator>(oldOp));
    newOp.recomputeSchema();
    newOp.computeDeliveredPhysicalProperties(context);
    context.computeAndSetTypeEnvironmentForOperator(newOp);
    AlgebricksConfig.ALGEBRICKS_LOGGER.finest(">>>> Structural properties for " + newOp.getPhysicalOperator() + ": "
            + newOp.getDeliveredPhysicalProperties() + "\n");

    PhysicalOptimizationsUtil.computeFDsAndEquivalenceClasses(newOp, context);
}

/**
 * Replace the original aggregate functions with their corresponding global aggregate function.
 */
protected void replaceOriginalAggFuncs(Set<SimilarAggregatesInfo> toReplaceSet) {
    for (SimilarAggregatesInfo sai : toReplaceSet) {
        for (AggregateExprInfo aei : sai.simAggs) {
            AbstractFunctionCallExpression afce = (AbstractFunctionCallExpression) aei.aggExprRef.getValue();
            afce.setFunctionInfo(aei.newFunInfo);
            afce.getArguments().clear();
            afce.getArguments().add(new MutableObject<ILogicalExpression>(sai.stepOneResult));
        }
    }
}

@Override
public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
        throws AlgebricksException {
    AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
    if (op.getOperatorTag() != LogicalOperatorTag.WRITE && op.getOperatorTag() != LogicalOperatorTag.WRITE_RESULT
            && op.getOperatorTag() != LogicalOperatorTag.DISTRIBUTE_RESULT) {
        return false;
    }
    if (!roots.contains(op)) {
        roots.add(new MutableObject<ILogicalOperator>(op));
    }
    return false;
}

protected LogicalVariable extractExprIntoAssignOpRef(ILogicalExpression gExpr, Mutable<ILogicalOperator> opRef2,
        IOptimizationContext context) throws AlgebricksException {
    LogicalVariable v = context.newVar();
    AssignOperator a = new AssignOperator(v, new MutableObject<ILogicalExpression>(gExpr));
    a.getInputs().add(new MutableObject<ILogicalOperator>(opRef2.getValue()));
    opRef2.setValue(a);
    if (gExpr.getExpressionTag() == LogicalExpressionTag.CONSTANT) {
        context.addNotToBeInlinedVar(v);
    }
    context.computeAndSetTypeEnvironmentForOperator(a);
    return v;
}

private boolean factorRedundantRhsVars(List<Pair<LogicalVariable, Mutable<ILogicalExpression>>> veList,
        Mutable<ILogicalOperator> opRef, Map<LogicalVariable, LogicalVariable> varRhsToLhs,
        IOptimizationContext context) throws AlgebricksException {
    varRhsToLhs.clear();
    ListIterator<Pair<LogicalVariable, Mutable<ILogicalExpression>>> iter = veList.listIterator();
    boolean changed = false;
    while (iter.hasNext()) {
        Pair<LogicalVariable, Mutable<ILogicalExpression>> p = iter.next();
        if (p.second.getValue().getExpressionTag() != LogicalExpressionTag.VARIABLE) {
            continue;
        }
        LogicalVariable v = GroupByOperator.getDecorVariable(p);
        LogicalVariable lhs = varRhsToLhs.get(v);
        if (lhs != null) {
            if (p.first != null) {
                AssignOperator assign = new AssignOperator(p.first, new MutableObject<ILogicalExpression>(
                        new VariableReferenceExpression(lhs)));
                ILogicalOperator op = opRef.getValue();
                assign.getInputs().add(new MutableObject<ILogicalOperator>(op));
                opRef.setValue(assign);
                context.computeAndSetTypeEnvironmentForOperator(assign);
            }
            iter.remove();
            changed = true;
        } else {
            varRhsToLhs.put(v, p.first);
        }
    }
    return changed;
}