/*
 * 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 io.prestosql.plugin.hive.metastore.glue;

import com.google.common.collect.ImmutableMap;
import io.airlift.slice.Slice;
import io.airlift.slice.Slices;
import io.prestosql.spi.predicate.Domain;
import io.prestosql.spi.predicate.Range;
import io.prestosql.spi.predicate.TupleDomain;
import io.prestosql.spi.predicate.ValueSet;
import io.prestosql.spi.type.BigintType;
import io.prestosql.spi.type.DateType;
import io.prestosql.spi.type.DecimalType;
import io.prestosql.spi.type.IntegerType;
import io.prestosql.spi.type.SmallintType;
import io.prestosql.spi.type.TinyintType;
import io.prestosql.spi.type.VarcharType;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.collect.ImmutableList.toImmutableList;

public class PartitionFilterBuilder
{
    public static final int DECIMAL_TYPE_PRECISION = 10;
    public static final int DECIMAL_TYPE_SCALE = 5;
    public static final DecimalType DECIMAL_TYPE = DecimalType.createDecimalType(DECIMAL_TYPE_PRECISION, DECIMAL_TYPE_SCALE);

    private final Map<String, Domain> domains = new HashMap<>();

    public PartitionFilterBuilder addStringValues(String columnName, String... values)
    {
        List<Slice> blockValues = Arrays.stream(values).map(Slices::utf8Slice).collect(toImmutableList());
        Domain domain = Domain.multipleValues(VarcharType.VARCHAR, blockValues);
        domains.compute(columnName, (k, v) -> v == null ? domain : v.union(domain));
        return this;
    }

    public PartitionFilterBuilder addBigintValues(String columnName, Long... values)
    {
        Domain domain = Domain.multipleValues(BigintType.BIGINT, Arrays.asList(values));
        domains.compute(columnName, (k, v) -> v == null ? domain : v.union(domain));
        return this;
    }

    public PartitionFilterBuilder addIntegerValues(String columnName, Long... values)
    {
        Domain domain = Domain.multipleValues(IntegerType.INTEGER, Arrays.asList(values));
        domains.compute(columnName, (k, v) -> v == null ? domain : v.union(domain));
        return this;
    }

    public PartitionFilterBuilder addSmallintValues(String columnName, Long... values)
    {
        Domain domain = Domain.multipleValues(SmallintType.SMALLINT, Arrays.asList(values));
        domains.compute(columnName, (k, v) -> v == null ? domain : v.union(domain));
        return this;
    }

    public PartitionFilterBuilder addTinyintValues(String columnName, Long... values)
    {
        Domain domain = Domain.multipleValues(TinyintType.TINYINT, Arrays.asList(values));
        domains.compute(columnName, (k, v) -> v == null ? domain : v.union(domain));
        return this;
    }

    public PartitionFilterBuilder addDecimalValues(String columnName, String... values)
    {
        checkArgument(values.length > 0);
        List<Long> encodedValues = Arrays.stream(values)
                .map(PartitionFilterBuilder::decimalOf)
                .collect(toImmutableList());
        Domain domain = Domain.multipleValues(DECIMAL_TYPE, encodedValues);
        domains.compute(columnName, (k, v) -> v == null ? domain : v.union(domain));
        return this;
    }

    public PartitionFilterBuilder addDateValues(String columnName, Long... values)
    {
        Domain domain = Domain.multipleValues(DateType.DATE, Arrays.asList(values));
        domains.compute(columnName, (k, v) -> v == null ? domain : v.union(domain));
        return this;
    }

    public PartitionFilterBuilder addRanges(String columnName, Range range, Range... ranges)
    {
        ValueSet values = ValueSet.ofRanges(range, ranges);
        Domain domain = Domain.create(values, false);
        domains.compute(columnName, (k, v) -> v == null ? domain : v.union(domain));
        return this;
    }

    public PartitionFilterBuilder addDomain(String columnName, Domain domain)
    {
        domains.compute(columnName, (k, v) -> v == null ? domain : v.union(domain));
        return this;
    }

    public TupleDomain<String> build()
    {
        return TupleDomain.withColumnDomains(ImmutableMap.copyOf(this.domains));
    }

    public static Long decimalOf(String value)
    {
        BigDecimal bigDecimalValue = new BigDecimal(value)
                .setScale(DECIMAL_TYPE_SCALE, RoundingMode.UP);
        return bigDecimalValue.unscaledValue().longValue();
    }
}

@@ -2105,38 +2105,36 @@
         domains.
-comput
+merg
 e(columnName, (k
@@ -2135,54 +2135,29 @@
 me, 
-(k, v) -%3E v == null ? domain : v.union(domain)
+domain, Domain::union
 );%0A 
@@ -2371,38 +2371,36 @@
         domains.
-comput
+merg
 e(columnName, (k
@@ -2401,54 +2401,29 @@
 me, 
-(k, v) -%3E v == null ? domain : v.union(domain)
+domain, Domain::union
 );%0A 
@@ -2640,38 +2640,36 @@
         domains.
-comput
+merg
 e(columnName, (k
@@ -2670,54 +2670,29 @@
 me, 
-(k, v) -%3E v == null ? domain : v.union(domain)
+domain, Domain::union
 );%0A 
@@ -2912,38 +2912,36 @@
         domains.
-comput
+merg
 e(columnName, (k
@@ -2942,54 +2942,29 @@
 me, 
-(k, v) -%3E v == null ? domain : v.union(domain)
+domain, Domain::union
 );%0A 
@@ -3181,38 +3181,36 @@
         domains.
-comput
+merg
 e(columnName, (k
@@ -3211,54 +3211,29 @@
 me, 
-(k, v) -%3E v == null ? domain : v.union(domain)
+domain, Domain::union
 );%0A 
@@ -3637,38 +3637,36 @@
         domains.
-comput
+merg
 e(columnName, (k
@@ -3667,54 +3667,29 @@
 me, 
-(k, v) -%3E v == null ? domain : v.union(domain)
+domain, Domain::union
 );%0A 
@@ -3897,38 +3897,36 @@
         domains.
-comput
+merg
 e(columnName, (k
@@ -3927,54 +3927,29 @@
 me, 
-(k, v) -%3E v == null ? domain : v.union(domain)
+domain, Domain::union
 );%0A 
@@ -4196,38 +4196,36 @@
         domains.
-comput
+merg
 e(columnName, (k
@@ -4226,54 +4226,29 @@
 me, 
-(k, v) -%3E v == null ? domain : v.union(domain)
+domain, Domain::union
 );%0A 
@@ -4378,14 +4378,12 @@
 ins.
-comput
+merg
 e(co
@@ -4396,54 +4396,29 @@
 me, 
-(k, v) -%3E v == null ? domain : v.union(domain)
+domain, Domain::union
 );%0A 

Replace Map.compute with Map.merge