/*
 * Licensed 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 com.facebook.presto.cost;

import com.facebook.presto.Session;
import com.facebook.presto.matching.Pattern;
import com.facebook.presto.metadata.Metadata;
import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.Constraint;
import com.facebook.presto.spi.predicate.TupleDomain;
import com.facebook.presto.spi.statistics.ColumnStatistics;
import com.facebook.presto.spi.statistics.TableStatistics;
import com.facebook.presto.spi.type.Type;
import com.facebook.presto.sql.planner.DomainTranslator;
import com.facebook.presto.sql.planner.Symbol;
import com.facebook.presto.sql.planner.iterative.Lookup;
import com.facebook.presto.sql.planner.plan.PlanNode;
import com.facebook.presto.sql.planner.plan.TableScanNode;
import com.facebook.presto.sql.tree.BooleanLiteral;
import com.facebook.presto.sql.tree.Expression;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalDouble;

import static com.facebook.presto.cost.StatsUtil.toStatsRepresentation;
import static com.facebook.presto.cost.SymbolStatsEstimate.UNKNOWN_STATS;
import static java.lang.Double.NEGATIVE_INFINITY;
import static java.lang.Double.POSITIVE_INFINITY;
import static java.util.Objects.requireNonNull;

public class TableScanStatsRule
        implements ComposableStatsCalculator.Rule
{
    private static final Pattern<TableScanNode> PATTERN = Pattern.typeOf(TableScanNode.class);

    private final Metadata metadata;

    public TableScanStatsRule(Metadata metadata)
    {
        this.metadata = requireNonNull(metadata, "metadata can not be null");
    }

    @Override
    public Pattern<TableScanNode> getPattern()
    {
        return PATTERN;
    }

    @Override
    public Optional<PlanNodeStatsEstimate> calculate(PlanNode node, StatsProvider sourceStats, Lookup lookup, Session session, Map<Symbol, Type> types)
    {
        TableScanNode tableScanNode = (TableScanNode) node;

        Constraint<ColumnHandle> constraint = getConstraint(tableScanNode, BooleanLiteral.TRUE_LITERAL, session, types);

        TableStatistics tableStatistics = metadata.getTableStatistics(session, tableScanNode.getTable(), constraint);
        Map<Symbol, SymbolStatsEstimate> outputSymbolStats = new HashMap<>();

        for (Map.Entry<Symbol, ColumnHandle> entry : tableScanNode.getAssignments().entrySet()) {
            Symbol symbol = entry.getKey();
            Type symbolType = types.get(symbol);
            Optional<ColumnStatistics> columnStatistics = Optional.ofNullable(tableStatistics.getColumnStatistics().get(entry.getValue()));
            outputSymbolStats.put(symbol, columnStatistics.map(statistics -> toSymbolStatistics(tableStatistics, statistics, session, symbolType)).orElse(UNKNOWN_STATS));
        }

        return Optional.of(PlanNodeStatsEstimate.builder()
                .setOutputRowCount(tableStatistics.getRowCount().getValue())
                .addSymbolStatistics(outputSymbolStats)
                .build());
    }

    private SymbolStatsEstimate toSymbolStatistics(TableStatistics tableStatistics, ColumnStatistics columnStatistics, Session session, Type type)
    {
        return SymbolStatsEstimate.builder()
                .setLowValue(asDouble(session, type, columnStatistics.getOnlyRangeColumnStatistics().getLowValue()).orElse(NEGATIVE_INFINITY))
                .setHighValue(asDouble(session, type, columnStatistics.getOnlyRangeColumnStatistics().getHighValue()).orElse(POSITIVE_INFINITY))
                .setNullsFraction(
                        columnStatistics.getNullsFraction().getValue()
                                / (columnStatistics.getNullsFraction().getValue() + columnStatistics.getOnlyRangeColumnStatistics().getFraction().getValue()))
                .setDistinctValuesCount(columnStatistics.getOnlyRangeColumnStatistics().getDistinctValuesCount().getValue())
                .setAverageRowSize(columnStatistics.getOnlyRangeColumnStatistics().getDataSize().getValue() / tableStatistics.getRowCount().getValue())
                .build();
    }

    private OptionalDouble asDouble(Session session, Type type, Optional<Object> optionalValue)
    {
        return optionalValue
                .map(value -> toStatsRepresentation(metadata, session, type, value))
                .orElseGet(OptionalDouble::empty);
    }

    private Constraint<ColumnHandle> getConstraint(TableScanNode node, Expression predicate, Session session, Map<Symbol, Type> types)
    {
        DomainTranslator.ExtractionResult decomposedPredicate = DomainTranslator.fromPredicate(
                metadata,
                session,
                predicate,
                types);

        TupleDomain<ColumnHandle> simplifiedConstraint = decomposedPredicate.getTupleDomain()
                .transform(node.getAssignments()::get)
                .intersect(node.getCurrentConstraint());

        // TODO Construct predicate like AddExchanges's LayoutConstraintEvaluator
        return new Constraint<>(simplifiedConstraint, bindings -> true);
    }
}

@@ -809,62 +809,8 @@
 nt;%0A
-import com.facebook.presto.spi.predicate.TupleDomain;%0A
 impo
@@ -857,32 +857,32 @@
 lumnStatistics;%0A
+
 import com.faceb
@@ -970,65 +970,8 @@
 pe;%0A
-import com.facebook.presto.sql.planner.DomainTranslator;%0A
 impo
@@ -1186,108 +1186,8 @@
 ode;
-%0Aimport com.facebook.presto.sql.tree.BooleanLiteral;%0Aimport com.facebook.presto.sql.tree.Expression;
 %0A%0Aim
@@ -2277,16 +2277,98 @@
  node;%0A%0A
+        // TODO Construct predicate like AddExchanges's LayoutConstraintEvaluator%0A
         
@@ -2405,19 +2405,20 @@
 raint = 
-get
+new 
 Constrai
@@ -2419,16 +2419,18 @@
 nstraint
+%3C%3E
 (tableSc
@@ -2439,53 +2439,49 @@
 Node
-, BooleanLiteral.TRUE_LITERAL, session, types
+.getCurrentConstraint(), bindings -%3E true
 );%0A%0A
@@ -4693,16 +4693,16 @@
 value))%0A
+
         
@@ -4753,716 +4753,7 @@
    %7D
-%0A%0A    private Constraint%3CColumnHandle%3E getConstraint(TableScanNode node, Expression predicate, Session session, Map%3CSymbol, Type%3E types)%0A    %7B%0A        DomainTranslator.ExtractionResult decomposedPredicate = DomainTranslator.fromPredicate(%0A                metadata,%0A                session,%0A                predicate,%0A                types);%0A%0A        TupleDomain%3CColumnHandle%3E simplifiedConstraint = decomposedPredicate.getTupleDomain()%0A                .transform(node.getAssignments()::get)%0A                .intersect(node.getCurrentConstraint());%0A%0A        // TODO Construct predicate like AddExchanges's LayoutConstraintEvaluator%0A        return new Constraint%3C%3E(simplifiedConstraint, bindings -%3E true);%0A    %7D
 %0A%7D%0A

Simplify Constraint construction in TableScanStatsRule