/*
 * 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.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.optimizer;

import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptPredicateList;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.plan.RelRule;
import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.rel.RelCollations;
import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.Calc;
import org.apache.calcite.rel.core.Filter;
import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.core.Project;
import org.apache.calcite.rel.core.Window;
import org.apache.calcite.rel.logical.LogicalWindow;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.rules.CoreRules;
import org.apache.calcite.rel.rules.SubstitutionRule;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexCorrelVariable;
import org.apache.calcite.rex.RexDynamicParam;
import org.apache.calcite.rex.RexExecutor;
import org.apache.calcite.rex.RexFieldAccess;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexLambda;
import org.apache.calcite.rex.RexLambdaRef;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexLocalRef;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexOver;
import org.apache.calcite.rex.RexProgram;
import org.apache.calcite.rex.RexProgramBuilder;
import org.apache.calcite.rex.RexRangeRef;
import org.apache.calcite.rex.RexShuttle;
import org.apache.calcite.rex.RexSimplify;
import org.apache.calcite.rex.RexSubQuery;
import org.apache.calcite.rex.RexUnknownAs;
import org.apache.calcite.rex.RexUtil;
import org.apache.calcite.rex.RexVisitorImpl;
import org.apache.calcite.rex.RexWindow;
import org.apache.calcite.rex.RexWindowBound;
import org.apache.calcite.sql.SqlAggFunction;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlOperator;
import org.apache.calcite.sql.fun.SqlRowOperator;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.tools.RelBuilder;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.Pair;
import org.apache.calcite.util.Util;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;

import org.checkerframework.checker.nullness.qual.Nullable;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Clone of Calcite {@link org.apache.calcite.rel.rules.ReduceExpressionsRule}
 * which does not fold ROW expressions.
 *
 * <P>Collection of planner rules that apply various simplifying transformations on
 * RexNode trees. Currently, there are two transformations:
 *
 * <ul>
 * <li>Constant reduction, which evaluates constant subtrees, replacing them
 * with a corresponding RexLiteral
 * <li>Removal of redundant casts, which occurs when the argument into the cast
 * is the same as the type of the resulting cast expression
 * </ul>
 *
 * @param <C> Configuration type
 */
@SuppressWarnings("ALL")
public abstract class ReduceExpressionsRule<C extends org.apache.calcite.rel.rules.ReduceExpressionsRule.Config>
        extends RelRule<C>
        implements SubstitutionRule {
    //~ Static fields/initializers ---------------------------------------------

    /**
     * Rule that reduces constants inside a {@link org.apache.calcite.rel.core.Filter}.
     * If the condition is a constant, the filter is removed (if TRUE) or replaced with
     * an empty {@link org.apache.calcite.rel.core.Values} (if FALSE or NULL).
     *
     * @see CoreRules#FILTER_REDUCE_EXPRESSIONS
     */
    public static class FilterReduceExpressionsRule
            extends org.apache.calcite.rel.rules.ReduceExpressionsRule<
            org.apache.calcite.rel.rules.ReduceExpressionsRule.FilterReduceExpressionsRule.FilterReduceExpressionsRuleConfig> {
        /** Creates a FilterReduceExpressionsRule. */
        protected FilterReduceExpressionsRule(FilterReduceExpressionsRule.FilterReduceExpressionsRuleConfig config) {
            super(config);
        }

        @Override public void onMatch(RelOptRuleCall call) {
            final Filter filter = call.rel(0);
            final List<RexNode> expList =
                    Lists.newArrayList(filter.getCondition());
            RexNode newConditionExp;
            boolean reduced;
            final RelMetadataQuery mq = call.getMetadataQuery();
            final RelOptPredicateList predicates =
                    mq.getPulledUpPredicates(filter.getInput());
            if (reduceExpressions(filter, expList, predicates, true,
                    config.matchNullability(), config.treatDynamicCallsAsConstant())) {
                assert expList.size() == 1;
                newConditionExp = expList.get(0);
                reduced = true;
            } else {
                // No reduction, but let's still test the original
                // predicate to see if it was already a constant,
                // in which case we don't need any runtime decision
                // about filtering.
                newConditionExp = filter.getCondition();
                reduced = false;
            }

            // Even if no reduction, let's still test the original
            // predicate to see if it was already a constant,
            // in which case we don't need any runtime decision
            // about filtering.
            if (newConditionExp.isAlwaysTrue()) {
                call.transformTo(
                        filter.getInput());
            } else if (newConditionExp instanceof RexLiteral
                    || RexUtil.isNullLiteral(newConditionExp, true)) {
                call.transformTo(createEmptyRelOrEquivalent(call, filter));
            } else if (reduced) {
                call.transformTo(call.builder()
                        .push(filter.getInput())
                        .filter(newConditionExp).build());
            } else {
                if (newConditionExp instanceof RexCall) {
                    boolean reverse = newConditionExp.getKind() == SqlKind.NOT;
                    if (reverse) {
                        newConditionExp = ((RexCall) newConditionExp).getOperands().get(0);
                    }
                    reduceNotNullableFilter(call, filter, newConditionExp, reverse);
                }
                return;
            }

            // New plan is absolutely better than old plan.
            call.getPlanner().prune(filter);
        }

        /**
         * For static schema systems, a filter that is always false or null can be
         * replaced by a values operator that produces no rows, as the schema
         * information can just be taken from the input Rel. In dynamic schema
         * environments, the filter might have an unknown input type, in these cases
         * they must define a system specific alternative to a Values operator, such
         * as inserting a limit 0 instead of a filter on top of the original input.
         *
         * <p>The default implementation of this method is to call
         * {@link RelBuilder#empty}, which for the static schema will be optimized
         * to an empty
         * {@link org.apache.calcite.rel.core.Values}.
         *
         * @param input rel to replace, assumes caller has already determined
         *              equivalence to Values operation for 0 records or a
         *              false filter.
         * @return equivalent but less expensive replacement rel
         */
        protected RelNode createEmptyRelOrEquivalent(RelOptRuleCall call, Filter input) {
            return call.builder().push(input).empty().build();
        }

        private void reduceNotNullableFilter(
                RelOptRuleCall call,
                Filter filter,
                RexNode rexNode,
                boolean reverse) {
            // If the expression is a IS [NOT] NULL on a non-nullable
            // column, then we can either remove the filter or replace
            // it with an Empty.
            boolean alwaysTrue;
            switch (rexNode.getKind()) {
                case IS_NULL:
                case IS_UNKNOWN:
                    alwaysTrue = false;
                    break;
                case IS_NOT_NULL:
                    alwaysTrue = true;
                    break;
                default:
                    return;
            }
            if (reverse) {
                alwaysTrue = !alwaysTrue;
            }
            RexNode operand = ((RexCall) rexNode).getOperands().get(0);
            if (operand instanceof RexInputRef) {
                RexInputRef inputRef = (RexInputRef) operand;
                if (!inputRef.getType().isNullable()) {
                    if (alwaysTrue) {
                        call.transformTo(filter.getInput());
                    } else {
                        call.transformTo(createEmptyRelOrEquivalent(call, filter));
                    }
                    // New plan is absolutely better than old plan.
                    call.getPlanner().prune(filter);
                }
            }
        }
    }

    /** Rule that reduces constants inside a
     * {@link org.apache.calcite.rel.core.Project}.
     *
     * @see CoreRules#PROJECT_REDUCE_EXPRESSIONS */
    public static class ProjectReduceExpressionsRule
            extends org.apache.calcite.rel.rules.ReduceExpressionsRule<
            org.apache.calcite.rel.rules.ReduceExpressionsRule.ProjectReduceExpressionsRule.ProjectReduceExpressionsRuleConfig> {
        /** Creates a ProjectReduceExpressionsRule. */
        protected ProjectReduceExpressionsRule(ProjectReduceExpressionsRule.ProjectReduceExpressionsRuleConfig config) {
            super(config);
        }

        @Override public void onMatch(RelOptRuleCall call) {
            final Project project = call.rel(0);
            final RelMetadataQuery mq = call.getMetadataQuery();
            final RelOptPredicateList predicates =
                    mq.getPulledUpPredicates(project.getInput());
            final List<RexNode> expList =
                    Lists.newArrayList(project.getProjects());
            if (ReduceExpressionsRule.reduceExpressions(project, expList, predicates, false,
                    config.matchNullability(), config.treatDynamicCallsAsConstant())) {
                assert !project.getProjects().equals(expList)
                        : "Reduced expressions should be different from original expressions";
                call.transformTo(
                        call.builder()
                                .push(project.getInput())
                                .project(expList, project.getRowType().getFieldNames())
                                .build());

                // New plan is absolutely better than old plan.
                call.getPlanner().prune(project);
            }
        }
    }

    /** Rule that reduces constants inside a {@link Join}.
     *
     * @see CoreRules#JOIN_REDUCE_EXPRESSIONS */
    public static class JoinReduceExpressionsRule
            extends org.apache.calcite.rel.rules.ReduceExpressionsRule<org.apache.calcite.rel.rules.ReduceExpressionsRule.JoinReduceExpressionsRule.JoinReduceExpressionsRuleConfig> {
        /** Creates a JoinReduceExpressionsRule. */
        protected JoinReduceExpressionsRule(JoinReduceExpressionsRule.JoinReduceExpressionsRuleConfig config) {
            super(config);
        }

        @Override public void onMatch(RelOptRuleCall call) {
            final Join join = call.rel(0);
            final List<RexNode> expList = Lists.newArrayList(join.getCondition());
            final int fieldCount = join.getLeft().getRowType().getFieldCount();
            final RelMetadataQuery mq = call.getMetadataQuery();
            final RelOptPredicateList leftPredicates =
                    mq.getPulledUpPredicates(join.getLeft());
            final RelOptPredicateList rightPredicates =
                    mq.getPulledUpPredicates(join.getRight());
            final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
            final RelOptPredicateList predicates =
                    leftPredicates.union(rexBuilder,
                            rightPredicates.shift(rexBuilder, fieldCount));
            if (!reduceExpressions(join, expList, predicates, true,
                    config.matchNullability(), config.treatDynamicCallsAsConstant())) {
                return;
            }
            call.transformTo(
                    join.copy(
                            join.getTraitSet(),
                            expList.get(0),
                            join.getLeft(),
                            join.getRight(),
                            join.getJoinType(),
                            join.isSemiJoinDone()));

            // New plan is absolutely better than old plan.
            call.getPlanner().prune(join);
        }
    }

    /**
     * Rule that reduces constants inside a {@link Calc}.
     *
     * @see CoreRules#CALC_REDUCE_EXPRESSIONS
     */
    public static class CalcReduceExpressionsRule
            extends org.apache.calcite.rel.rules.ReduceExpressionsRule<org.apache.calcite.rel.rules.ReduceExpressionsRule.CalcReduceExpressionsRule.CalcReduceExpressionsRuleConfig> {
        /**
         * Creates a CalcReduceExpressionsRule.
         */
        protected CalcReduceExpressionsRule(CalcReduceExpressionsRule.CalcReduceExpressionsRuleConfig config) {
            super(config);
        }

        @Override
        public void onMatch(RelOptRuleCall call) {
            Calc calc = call.rel(0);
            RexProgram program = calc.getProgram();
            final List<RexNode> exprList = program.getExprList();

            // Form a list of expressions with sub-expressions fully expanded.
            final List<RexNode> expandedExprList = new ArrayList<>();
            final RexShuttle shuttle =
                    new RexShuttle() {
                        @Override
                        public RexNode visitLocalRef(RexLocalRef localRef) {
                            return expandedExprList.get(localRef.getIndex());
                        }
                    };
            for (RexNode expr : exprList) {
                expandedExprList.add(expr.accept(shuttle));
            }
            final RelOptPredicateList predicates = RelOptPredicateList.EMPTY;
            if (reduceExpressions(calc, expandedExprList, predicates, false,
                    config.matchNullability(), config.treatDynamicCallsAsConstant())) {
                final RexProgramBuilder builder =
                        new RexProgramBuilder(
                                calc.getInput().getRowType(),
                                calc.getCluster().getRexBuilder());
                final List<RexLocalRef> list = new ArrayList<>();
                for (RexNode expr : expandedExprList) {
                    list.add(builder.registerInput(expr));
                }
                if (program.getCondition() != null) {
                    final int conditionIndex =
                            program.getCondition().getIndex();
                    final RexNode newConditionExp =
                            expandedExprList.get(conditionIndex);
                    if (newConditionExp.isAlwaysTrue()) {
                        // condition is always TRUE - drop it.
                    } else if (newConditionExp instanceof RexLiteral
                            || RexUtil.isNullLiteral(newConditionExp, true)) {
                        // condition is always NULL or FALSE - replace calc
                        // with empty.
                        call.transformTo(createEmptyRelOrEquivalent(call, calc));
                        return;
                    } else {
                        builder.addCondition(list.get(conditionIndex));
                    }
                }
                int k = 0;
                for (RexLocalRef projectExpr : program.getProjectList()) {
                    final int index = projectExpr.getIndex();
                    builder.addProject(
                            list.get(index).getIndex(),
                            program.getOutputRowType().getFieldNames().get(k++));
                }
                call.transformTo(
                        calc.copy(calc.getTraitSet(), calc.getInput(), builder.getProgram()));

                // New plan is absolutely better than old plan.
                call.getPlanner().prune(calc);
            }
        }

        /**
         * For static schema systems, a filter that is always false or null can be
         * replaced by a values operator that produces no rows, as the schema
         * information can just be taken from the input Rel. In dynamic schema
         * environments, the filter might have an unknown input type, in these cases
         * they must define a system specific alternative to a Values operator, such
         * as inserting a limit 0 instead of a filter on top of the original input.
         *
         * <p>The default implementation of this method is to call
         * {@link RelBuilder#empty}, which for the static schema will be optimized
         * to an Immutable.Config.of()
         * {@link org.apache.calcite.rel.core.Values}.
         *
         * @param input rel to replace, assumes caller has already determined
         *              equivalence to Values operation for 0 records or a
         *              false filter.
         * @return equivalent but less expensive replacement rel
         */
        protected RelNode createEmptyRelOrEquivalent(RelOptRuleCall call, Calc input) {
            return call.builder().push(input).empty().build();
        }
    }

    /** Rule that reduces constants inside a {@link Window}.
     *
     * @see CoreRules#WINDOW_REDUCE_EXPRESSIONS */
    public static class WindowReduceExpressionsRule
            extends org.apache.calcite.rel.rules.ReduceExpressionsRule<org.apache.calcite.rel.rules.ReduceExpressionsRule.WindowReduceExpressionsRule.WindowReduceExpressionsRuleConfig> {
        /** Creates a WindowReduceExpressionsRule. */
        protected WindowReduceExpressionsRule(WindowReduceExpressionsRule.WindowReduceExpressionsRuleConfig config) {
            super(config);
        }

        @Override public void onMatch(RelOptRuleCall call) {
            LogicalWindow window = call.rel(0);
            RexBuilder rexBuilder = window.getCluster().getRexBuilder();
            final RelMetadataQuery mq = call.getMetadataQuery();
            final RelOptPredicateList predicates = mq
                    .getPulledUpPredicates(window.getInput());

            boolean reduced = false;
            final List<Window.Group> groups = new ArrayList<>();
            for (Window.Group group : window.groups) {
                List<Window.RexWinAggCall> aggCalls = new ArrayList<>();
                for (Window.RexWinAggCall aggCall : group.aggCalls) {
                    final List<RexNode> expList = new ArrayList<>(aggCall.getOperands());
                    if (reduceExpressions(window, expList, predicates)) {
                        aggCall =
                                new Window.RexWinAggCall(aggCall.getParserPosition(),
                                        (SqlAggFunction) aggCall.getOperator(),
                                        aggCall.type, expList,
                                        aggCall.ordinal, aggCall.distinct, aggCall.ignoreNulls);
                        reduced = true;
                    }
                    aggCalls.add(aggCall);
                }

                final ImmutableBitSet.Builder keyBuilder = ImmutableBitSet.builder();
                for (Integer key : group.keys) {
                    if (!predicates.constantMap.containsKey(
                            rexBuilder.makeInputRef(window.getInput(), key))) {
                        keyBuilder.set(key);
                    }
                }
                final ImmutableBitSet keys = keyBuilder.build();
                reduced |= keys.cardinality() != group.keys.cardinality();

                final List<RelFieldCollation> collationsList = group.orderKeys
                        .getFieldCollations().stream()
                        .filter(fc ->
                                !predicates.constantMap.containsKey(
                                        rexBuilder.makeInputRef(window.getInput(),
                                                fc.getFieldIndex())))
                        .collect(Collectors.toList());

                boolean collationReduced =
                        group.orderKeys.getFieldCollations().size() != collationsList.size();
                reduced |= collationReduced;
                RelCollation relCollation = collationReduced
                        ? RelCollations.of(collationsList)
                        : group.orderKeys;
                groups.add(
                        new Window.Group(keys, group.isRows, group.lowerBound,
                                group.upperBound, group.exclude, relCollation, aggCalls));
            }
            if (reduced) {
                call.transformTo(LogicalWindow
                        .create(window.getTraitSet(), window.getHints(), window.getInput(),
                                window.getConstants(), window.getRowType(), groups));
                call.getPlanner().prune(window);
            }
        }
    }

    //~ Constructors -----------------------------------------------------------

    /** Creates a ReduceExpressionsRule. */
    protected ReduceExpressionsRule(C config) {
        super(config);
    }

    //~ Methods ----------------------------------------------------------------

    /**
     * Reduces a list of expressions.
     *
     * <p>The {@code matchNullability} flag comes into play when reducing an
     * expression whose type is nullable. Suppose we are reducing an expression
     * {@code CASE WHEN 'a' = 'a' THEN 1 ELSE NULL END}. Before reduction the
     * type is {@code INTEGER} (nullable), but after reduction the literal 1 has
     * type {@code INTEGER NOT NULL}.
     *
     * <p>In some situations it is more important to preserve types; in this
     * case you should use {@code matchNullability = true} (which used to be
     * the default behavior of this method), and it will cast the literal to
     * {@code INTEGER} (nullable).
     *
     * <p>In other situations, you would rather propagate the new stronger type,
     * because it may allow further optimizations later; pass
     * {@code matchNullability = false} and no cast will be added, but you may
     * need to adjust types elsewhere in the expression tree.
     *
     * @param rel     Relational expression
     * @param expList List of expressions, modified in place
     * @param predicates Constraints known to hold on input expressions
     * @param unknownAsFalse Whether UNKNOWN will be treated as FALSE
     * @param matchNullability Whether Calcite should add a CAST to a literal
     *                         resulting from simplification and expression if the
     *                         expression had nullable type and the literal is
     *                         NOT NULL
     * @param treatDynamicCallsAsConstant Whether to treat dynamic functions as
     *                                    constants
     *
     * @return whether reduction found something to change, and succeeded
     */
    protected static boolean reduceExpressions(RelNode rel, List<RexNode> expList,
                                               RelOptPredicateList predicates, boolean unknownAsFalse,
                                               boolean matchNullability, boolean treatDynamicCallsAsConstant) {
        final RelOptCluster cluster = rel.getCluster();
        final RexBuilder rexBuilder = cluster.getRexBuilder();
        final List<RexNode> originExpList = Lists.newArrayList(expList);
        final RexExecutor executor =
                Util.first(cluster.getPlanner().getExecutor(), RexUtil.EXECUTOR);
        final RexSimplify simplify =
                new RexSimplify(rexBuilder, predicates, executor);

        // Simplify predicates in place
        final RexUnknownAs unknownAs = RexUnknownAs.falseIf(unknownAsFalse);
        final boolean reduced =
                reduceExpressionsInternal(rel, simplify, unknownAs,
                        expList, predicates, treatDynamicCallsAsConstant);

        boolean simplified = false;
        /*
        Some Calcite simplifications are wrong, in particular the ones producing TIMESTAMP results.
        for (int i = 0; i < expList.size(); i++) {
            final RexNode expr2 =
                    simplify.simplifyPreservingType(expList.get(i), unknownAs,
                            matchNullability);
            if (!expr2.equals(expList.get(i))) {
                expList.set(i, expr2);
                simplified = true;
            }
        }
         */

        if (reduced && simplified) {
            return !originExpList.equals(expList);
        }

        return reduced || simplified;
    }

    protected static boolean reduceExpressionsInternal(RelNode rel,
                                                       RexSimplify simplify, RexUnknownAs unknownAs, List<RexNode> expList,
                                                       RelOptPredicateList predicates, boolean treatDynamicCallsAsConstant) {
        // Replace predicates on CASE to CASE on predicates.
        boolean changed = new ReduceExpressionsRule.CaseShuttle().mutate(expList);
        // Apply our custom rewrite rules
        changed = changed || new RexOptimize(simplify.rexBuilder).mutate(expList);

        // Find reducible expressions.
        final List<RexNode> constExps = new ArrayList<>();
        List<Boolean> addCasts = new ArrayList<>();
        findReducibleExps(rel.getCluster().getTypeFactory(), expList,
                predicates.constantMap, constExps, addCasts, treatDynamicCallsAsConstant);
        if (constExps.isEmpty()) {
            return changed;
        }

        final List<RexNode> constExps2 = Lists.newArrayList(constExps);
        if (!predicates.constantMap.isEmpty()) {
            final List<Map.Entry<RexNode, RexNode>> pairs =
                    Lists.newArrayList(predicates.constantMap.entrySet());
            RexReplacer replacer =
                    new RexReplacer(simplify, unknownAs, Pair.left(pairs),
                            Pair.right(pairs), Collections.nCopies(pairs.size(), false));
            replacer.mutate(constExps2);
        }

        // Compute the values they reduce to.
        RexExecutor executor = rel.getCluster().getPlanner().getExecutor();
        if (executor == null) {
            // Cannot reduce expressions: caller has not set an executor in their
            // environment. Caller should execute something like the following before
            // invoking the planner:
            //
            // final RexExecutorImpl executor =
            //   new RexExecutorImpl(Schemas.createDataContext(null));
            // rootRel.getCluster().getPlanner().setExecutor(executor);
            return changed;
        }

        final List<RexNode> reducedValues = new ArrayList<>();
        executor.reduce(simplify.rexBuilder, constExps2, reducedValues);

        // Use RexNode.digest to judge whether each newly generated RexNode
        // is equivalent to the original one.
        if (RexUtil.strings(constExps).equals(RexUtil.strings(reducedValues))) {
            return changed;
        }

        // For Project, we have to be sure to preserve the result
        // types, so always cast regardless of the expression type.
        // For other RelNodes like Filter, in general, this isn't necessary,
        // and the presence of casts could hinder other rules such as sarg
        // analysis, which require bare literals.  But there are special cases,
        // like when the expression is a UDR argument, that need to be
        // handled as special cases.
        if (rel instanceof Project) {
            addCasts = Collections.nCopies(reducedValues.size(), true);
        }

        new RexReplacer(simplify, unknownAs, constExps, reducedValues, addCasts)
                .mutate(expList);
        return true;
    }

    /**
     * Locates expressions that can be reduced to literals or converted to
     * expressions with redundant casts removed.
     *
     * @param typeFactory    Type factory
     * @param exps           list of candidate expressions to be examined for
     *                       reduction
     * @param constants      List of expressions known to be constant
     * @param constExps      returns the list of expressions that can be constant
     *                       reduced
     * @param addCasts       indicator for each expression that can be constant
     *                       reduced, whether a cast of the resulting reduced
     *                       expression is potentially necessary
     * @param treatDynamicCallsAsConstant Whether to treat dynamic functions as
     *                                    constants
     */
    protected static void findReducibleExps(RelDataTypeFactory typeFactory,
                                            List<RexNode> exps, ImmutableMap<RexNode, RexNode> constants,
                                            List<RexNode> constExps, List<Boolean> addCasts, boolean treatDynamicCallsAsConstant) {
        ReducibleExprLocator gardener =
                new ReducibleExprLocator(typeFactory, constants, constExps,
                        addCasts, treatDynamicCallsAsConstant);
        for (RexNode exp : exps) {
            gardener.analyze(exp);
        }
        assert constExps.size() == addCasts.size();
    }

    /** Pushes predicates into a CASE.
     *
     * <p>We have a loose definition of 'predicate': any boolean expression will
     * do, except CASE. For example '(CASE ...) = 5' or '(CASE ...) IS NULL'.
     */
    public static RexCall pushPredicateIntoCase(RexCall call) {
        if (call.getType().getSqlTypeName() != SqlTypeName.BOOLEAN) {
            return call;
        }
        switch (call.getKind()) {
            case CASE:
            case AND:
            case OR:
                return call; // don't push CASE into CASE!
            case EQUALS: {
                // checks that the EQUALS operands may be split and
                // doesn't push EQUALS into CASE
                List<RexNode> equalsOperands = call.getOperands();
                ImmutableBitSet left = RelOptUtil.InputFinder.bits(equalsOperands.get(0));
                ImmutableBitSet right = RelOptUtil.InputFinder.bits(equalsOperands.get(1));
                if (!left.isEmpty() && !right.isEmpty() && left.intersect(right).isEmpty()) {
                    return call;
                }
                break;
            }
            default:
                break;
        }
        int caseOrdinal = -1;
        final List<RexNode> operands = call.getOperands();
        for (int i = 0; i < operands.size(); i++) {
            RexNode operand = operands.get(i);
            if (operand.getKind() == SqlKind.CASE) {
                caseOrdinal = i;
            }
        }
        if (caseOrdinal < 0) {
            return call;
        }
        // Convert
        //   f(CASE WHEN p1 THEN v1 ... END, arg)
        // to
        //   CASE WHEN p1 THEN f(v1, arg) ... END
        final RexCall case_ = (RexCall) operands.get(caseOrdinal);
        final List<RexNode> nodes = new ArrayList<>();
        for (int i = 0; i < case_.getOperands().size(); i++) {
            RexNode node = case_.getOperands().get(i);
            if (!RexUtil.isCasePredicate(case_, i)) {
                node = substitute(call, caseOrdinal, node);
            }
            nodes.add(node);
        }
        return case_.clone(call.getType(), nodes);
    }

    /** Converts op(arg0, ..., argOrdinal, ..., argN) to op(arg0,..., node, ..., argN). */
    protected static RexNode substitute(RexCall call, int ordinal, RexNode node) {
        final List<RexNode> newOperands = Lists.newArrayList(call.getOperands());
        newOperands.set(ordinal, node);
        return call.clone(call.getType(), newOperands);
    }

    //~ Inner Classes ----------------------------------------------------------

    /**
     * Replaces expressions with their reductions. Note that we only have to
     * look for RexCall, since nothing else is reducible in the first place.
     */
    protected static class RexReplacer extends RexShuttle {
        private final RexSimplify simplify;
        private final List<RexNode> reducibleExps;
        private final List<RexNode> reducedValues;
        private final List<Boolean> addCasts;

        RexReplacer(
                RexSimplify simplify,
                RexUnknownAs unknownAs,
                List<RexNode> reducibleExps,
                List<RexNode> reducedValues,
                List<Boolean> addCasts) {
            this.simplify = simplify;
            this.reducibleExps = reducibleExps;
            this.reducedValues = reducedValues;
            this.addCasts = addCasts;
        }

        @Override public RexNode visitInputRef(RexInputRef inputRef) {
            RexNode node = visit(inputRef);
            if (node == null) {
                return super.visitInputRef(inputRef);
            }
            return node;
        }

        @Override public RexNode visitCall(RexCall call) {
            RexNode node = visit(call);
            if (node != null) {
                return node;
            }
            node = super.visitCall(call);
            return node;
        }

        private @Nullable RexNode visit(final RexNode call) {
            int i = reducibleExps.indexOf(call);
            if (i == -1) {
                return null;
            }
            RexNode replacement = reducedValues.get(i);
            if (addCasts.get(i)
                    && (replacement.getType() != call.getType())) {
                // Handle change from nullable to NOT NULL by claiming
                // that the result is still nullable, even though
                // we know it isn't.
                //
                // Also, we cannot reduce CAST('abc' AS VARCHAR(4)) to 'abc'.
                // If we make 'abc' of type VARCHAR(4), we may later encounter
                // the same expression in a Project's digest where it has
                // type VARCHAR(3), and that's wrong.
                RelDataType type = call.getType();
                replacement = simplify.rexBuilder.makeAbstractCast(type, replacement, false);
            }
            return replacement;
        }
    }

    /**
     * Helper class used to locate expressions that either can be reduced to
     * literals or contain redundant casts.
     */
    protected static class ReducibleExprLocator extends RexVisitorImpl<Void> {
        /** Whether an expression is constant, and if so, whether it can be
         * reduced to a simpler constant. */
        enum Constancy {
            NON_CONSTANT, REDUCIBLE_CONSTANT, IRREDUCIBLE_CONSTANT
        }

        private final boolean treatDynamicCallsAsConstant;

        private final List<ReducibleExprLocator.Constancy> stack = new ArrayList<>();

        private final ImmutableMap<RexNode, RexNode> constants;

        private final List<RexNode> constExprs;

        private final List<Boolean> addCasts;

        private final Deque<SqlOperator> parentCallTypeStack = new ArrayDeque<>();

        ReducibleExprLocator(RelDataTypeFactory typeFactory,
                             ImmutableMap<RexNode, RexNode> constants, List<RexNode> constExprs,
                             List<Boolean> addCasts, boolean treatDynamicCallsAsConstant) {
            // go deep
            super(true);
            this.constants = constants;
            this.constExprs = constExprs;
            this.addCasts = addCasts;
            this.treatDynamicCallsAsConstant = treatDynamicCallsAsConstant;
        }

        public void analyze(RexNode exp) {
            assert stack.isEmpty();

            exp.accept(this);

            // Deal with top of stack
            assert stack.size() == 1;
            assert parentCallTypeStack.isEmpty();
            ReducibleExprLocator.Constancy rootConstancy = stack.get(0);
            if (rootConstancy == ReducibleExprLocator.Constancy.REDUCIBLE_CONSTANT) {
                // The entire subtree was constant, so add it to the result.
                addResult(exp);
            }
            stack.clear();
        }

        private Void pushVariable() {
            stack.add(ReducibleExprLocator.Constancy.NON_CONSTANT);
            return null;
        }

        private void addResult(RexNode exp) {
            // Cast of literal can't be reduced, so skip those (otherwise we'd
            // go into an infinite loop as we add them back).
            if (exp.getKind() == SqlKind.CAST) {
                RexCall cast = (RexCall) exp;
                RexNode operand = cast.getOperands().get(0);
                if (operand instanceof RexLiteral) {
                    return;
                }
                if (operand instanceof RexCall) {
                    RexCall opCall = (RexCall) operand;
                    if (opCall.getKind() == SqlKind.ARRAY_VALUE_CONSTRUCTOR) {
                        // We can't simplify casts of arrays even if arrays are constant.
                        return;
                    }
                }
            }
            constExprs.add(exp);

            // In the case where the expression corresponds to a UDR argument,
            // we need to preserve casts.  Note that this only applies to
            // the topmost argument, not expressions nested within the UDR
            // call.
            //
            // REVIEW zfong 6/13/08 - Are there other expressions where we
            // also need to preserve casts?
            SqlOperator op = parentCallTypeStack.peek();
            if (op == null) {
                addCasts.add(false);
            } else {
                addCasts.add(isUdf(op));
            }
        }

        private static Boolean isUdf(@SuppressWarnings("unused") SqlOperator operator) {
            // return operator instanceof UserDefinedRoutine
            return false;
        }

        @Override public Void visitInputRef(RexInputRef inputRef) {
            final RexNode constant = constants.get(inputRef);
            if (constant != null) {
                if (constant instanceof RexCall || constant instanceof RexDynamicParam) {
                    constant.accept(this);
                } else {
                    stack.add(ReducibleExprLocator.Constancy.REDUCIBLE_CONSTANT);
                }
                return null;
            }
            return pushVariable();
        }

        @Override public Void visitLiteral(RexLiteral literal) {
            stack.add(ReducibleExprLocator.Constancy.IRREDUCIBLE_CONSTANT);
            return null;
        }

        void processWindowBound(RexWindowBound bound) {
            RexNode offset = bound.getOffset();
            if (offset == null) {
                return;
            }
            bound.accept(this);
            ReducibleExprLocator.Constancy constancy = Util.last(stack);
            if (constancy == ReducibleExprLocator.Constancy.REDUCIBLE_CONSTANT) {
                addResult(offset);
            }
            Util.last(stack, 1).clear();
        }

        @Override public Void visitOver(RexOver over) {
            // assume non-constant (running SUM(1) looks constant but isn't)
            analyzeCall(over, ReducibleExprLocator.Constancy.NON_CONSTANT);
            final RexWindow window = over.getWindow();
            this.processWindowBound(window.getLowerBound());
            this.processWindowBound(window.getUpperBound());
            return null;
        }

        @Override public Void visitCorrelVariable(RexCorrelVariable variable) {
            return pushVariable();
        }

        @Override public Void visitCall(RexCall call) {
            // assume REDUCIBLE_CONSTANT until proven otherwise
            analyzeCall(call, ReducibleExprLocator.Constancy.REDUCIBLE_CONSTANT);
            return null;
        }

        @Override public Void visitSubQuery(RexSubQuery subQuery) {
            analyzeCall(subQuery, ReducibleExprLocator.Constancy.REDUCIBLE_CONSTANT);
            return null;
        }

        private void analyzeCall(RexCall call, ReducibleExprLocator.Constancy callConstancy) {
            parentCallTypeStack.push(call.getOperator());

            // visit operands, pushing their states onto stack
            super.visitCall(call);

            // look for NON_CONSTANT operands
            int operandCount = call.getOperands().size();
            List<ReducibleExprLocator.Constancy> operandStack = Util.last(stack, operandCount);
            for (ReducibleExprLocator.Constancy operandConstancy : operandStack) {
                if (operandConstancy == ReducibleExprLocator.Constancy.NON_CONSTANT) {
                    callConstancy = ReducibleExprLocator.Constancy.NON_CONSTANT;
                    break;
                }
            }

            // Even if all operands are constant, the call itself may
            // be non-deterministic.
            if (!call.getOperator().isDeterministic()) {
                callConstancy = ReducibleExprLocator.Constancy.NON_CONSTANT;
            } else if (!treatDynamicCallsAsConstant
                    && call.getOperator().isDynamicFunction()) {
                // In some circumstances, we should avoid caching the plan if we have dynamic functions.
                // If desired, treat this situation the same as a non-deterministic function.
                callConstancy = ReducibleExprLocator.Constancy.NON_CONSTANT;
            }

            // Row operator itself can't be reduced to a literal, but if
            // the operands are constants, we still want to reduce those.
            // This part was removed in Calcite, but we want to prevent optimizing
            // IS_NULL(ROW(NULL)) to NULL, so we avoid classifying ROW as a constant ever.
            if ((callConstancy == ReducibleExprLocator.Constancy.REDUCIBLE_CONSTANT)
                    && (call.getOperator() instanceof SqlRowOperator)) {
                callConstancy = ReducibleExprLocator.Constancy.NON_CONSTANT;
            }
            // https://github.com/feldera/feldera/issues/4700:
            // Disable all expression evaluations for TIME and TIMESTAMP values, since they are broken in Calcite:
            // Calcite does not support TIMESTAMP with precisions above 3
            SqlTypeName resultType = call.getType().getSqlTypeName();
            if (resultType == SqlTypeName.TIMESTAMP ||
                    resultType == SqlTypeName.TIME ||
                    SqlTypeName.INTERVAL_TYPES.contains(resultType))
                callConstancy = Constancy.NON_CONSTANT;
            for (int iOperand = 0; iOperand < operandCount; ++iOperand) {
                RexNode operand = call.getOperands().get(iOperand);
                SqlTypeName operandType = operand.getType().getSqlTypeName();
                if (operandType == SqlTypeName.TIMESTAMP || operandType == SqlTypeName.TIME) {
                    callConstancy = Constancy.NON_CONSTANT;
                }
            }

            if (callConstancy == ReducibleExprLocator.Constancy.NON_CONSTANT) {
                // any REDUCIBLE_CONSTANT children are now known to be maximal
                // reducible subtrees, so they can be added to the result
                // list
                for (int iOperand = 0; iOperand < operandCount; ++iOperand) {
                    ReducibleExprLocator.Constancy constancy = operandStack.get(iOperand);
                    if (constancy == ReducibleExprLocator.Constancy.REDUCIBLE_CONSTANT) {
                        addResult(call.getOperands().get(iOperand));
                    }
                }
            }

            // pop operands off of the stack
            operandStack.clear();

            // pop this parent call operator off the stack
            parentCallTypeStack.pop();

            // push constancy result for this call onto stack
            stack.add(callConstancy);
        }

        @Override public Void visitDynamicParam(RexDynamicParam dynamicParam) {
            return pushVariable();
        }

        @Override public Void visitRangeRef(RexRangeRef rangeRef) {
            return pushVariable();
        }

        @Override public Void visitFieldAccess(RexFieldAccess fieldAccess) {
            return pushVariable();
        }

        @Override public Void visitLambda(RexLambda lambda) {
            return pushVariable();
        }

        @Override public Void visitLambdaRef(RexLambdaRef lambda) {
            return pushVariable();
        }
    }

    /** Shuttle that pushes predicates into a CASE. */
    protected static class CaseShuttle extends RexShuttle {
        @Override public RexNode visitCall(RexCall call) {
            for (;;) {
                call = (RexCall) super.visitCall(call);
                final RexCall old = call;
                call = pushPredicateIntoCase(call);
                if (call == old) {
                    return call;
                }
            }
        }
    }

    /** Change the builder factory for a rule to not simplify */
    static <C extends RelRule.Config> C noSimplify(C c, Class<C> clazz) {
        return c.withRelBuilderFactory(RelBuilder.proto(RelBuilder.Config.DEFAULT.withSimplify(false))).as(clazz);
    }

    public static final FilterReduceExpressionsRule FILTER_REDUCE_EXPRESSIONS =
            new FilterReduceExpressionsRule(
                    noSimplify(
                            org.apache.calcite.rel.rules.ReduceExpressionsRule.FilterReduceExpressionsRule.FilterReduceExpressionsRuleConfig.DEFAULT,
                            org.apache.calcite.rel.rules.ReduceExpressionsRule.FilterReduceExpressionsRule.FilterReduceExpressionsRuleConfig.class));

    public static final ProjectReduceExpressionsRule PROJECT_REDUCE_EXPRESSIONS =
            new ProjectReduceExpressionsRule(
                    noSimplify(
                            org.apache.calcite.rel.rules.ReduceExpressionsRule.ProjectReduceExpressionsRule.ProjectReduceExpressionsRuleConfig.DEFAULT,
                            org.apache.calcite.rel.rules.ReduceExpressionsRule.ProjectReduceExpressionsRule.ProjectReduceExpressionsRuleConfig.class));

    public static final JoinReduceExpressionsRule JOIN_REDUCE_EXPRESSIONS =
            new JoinReduceExpressionsRule(
                    noSimplify(
                            org.apache.calcite.rel.rules.ReduceExpressionsRule.JoinReduceExpressionsRule.JoinReduceExpressionsRuleConfig.DEFAULT,
                            org.apache.calcite.rel.rules.ReduceExpressionsRule.JoinReduceExpressionsRule.JoinReduceExpressionsRuleConfig.class));

    public static final WindowReduceExpressionsRule WINDOW_REDUCE_EXPRESSIONS =
            new WindowReduceExpressionsRule(
                    noSimplify(
                            org.apache.calcite.rel.rules.ReduceExpressionsRule.WindowReduceExpressionsRule.WindowReduceExpressionsRuleConfig.DEFAULT,
                            org.apache.calcite.rel.rules.ReduceExpressionsRule.WindowReduceExpressionsRule.WindowReduceExpressionsRuleConfig.class));

    public static final CalcReduceExpressionsRule CALC_REDUCE_EXPRESSIONS =
            new CalcReduceExpressionsRule(
                    noSimplify(
                            org.apache.calcite.rel.rules.ReduceExpressionsRule.CalcReduceExpressionsRule.CalcReduceExpressionsRuleConfig.DEFAULT,
                            org.apache.calcite.rel.rules.ReduceExpressionsRule.CalcReduceExpressionsRule.CalcReduceExpressionsRuleConfig.class));
}
