content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.beam.sdk.util;
  
  import com.fasterxml.jackson.databind.JsonNode;
  import com.google.auto.value.AutoValue;
  import java.math.BigDecimal;
  import java.time.LocalDate;
  import java.time.LocalDateTime;
  import java.time.LocalTime;
  import java.util.function.Function;
  import java.util.function.Predicate;
  import org.apache.beam.sdk.util.RowJson.UnsupportedRowJsonException;
  import org.joda.time.DateTime;
  
  /**
   * Contains utilities for extracting primitive values from JSON nodes.
   *
   * <p>Performs validation and rejects values which are out of bounds.
   */
  class RowJsonValueExtractors {
  
    public interface ValueExtractor<V> {
      V extractValue(JsonNode value);
    }
  
    /**
     * Extracts byte value from JsonNode if it is within bounds.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<Byte> byteValueExtractor() {
      return ValidatingValueExtractor.<Byte>builder()
          .setExtractor(jsonNode -> (byte) jsonNode.intValue())
          .setValidator(
              jsonNode ->
                  jsonNode.isIntegralNumber()
                      && jsonNode.canConvertToInt()
                      && jsonNode.intValue() >= Byte.MIN_VALUE
                      && jsonNode.intValue() <= Byte.MAX_VALUE)
          .build();
    }
  
    /**
     * Extracts short value from JsonNode if it is within bounds.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<Short> shortValueExtractor() {
      return ValidatingValueExtractor.<Short>builder()
          .setExtractor(jsonNode -> (short) jsonNode.intValue())
          .setValidator(
              jsonNode ->
                  jsonNode.isIntegralNumber()
                      && jsonNode.canConvertToInt()
                      && jsonNode.intValue() >= Short.MIN_VALUE
                      && jsonNode.intValue() <= Short.MAX_VALUE)
          .build();
    }
  
    /**
     * Extracts int value from JsonNode if it is within bounds.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<Integer> intValueExtractor() {
      return ValidatingValueExtractor.<Integer>builder()
          .setExtractor(JsonNode::intValue)
          .setValidator(jsonNode -> jsonNode.isIntegralNumber() && jsonNode.canConvertToInt())
          .build();
    }
  
    /**
     * Extracts long value from JsonNode if it is within bounds.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<Long> longValueExtractor() {
      return ValidatingValueExtractor.<Long>builder()
          .setExtractor(JsonNode::longValue)
          .setValidator(jsonNode -> jsonNode.isIntegralNumber() && jsonNode.canConvertToLong())
          .build();
    }
  
    /**
     * Extracts float value from JsonNode if it is within bounds.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<Float> floatValueExtractor() {
      return ValidatingValueExtractor.<Float>builder()
          .setExtractor(JsonNode::floatValue)
          .setValidator(
              jsonNode ->
                  jsonNode.isFloat()
  
                      // Either floating number which allows lossless conversion to float
                      || (jsonNode.isFloatingPointNumber()
                          && jsonNode.doubleValue() == (double) (float) jsonNode.doubleValue())
  
                      // Or an integer number which allows lossless conversion to float
                      || (jsonNode.isIntegralNumber()
                          && jsonNode.canConvertToInt()
                          && jsonNode.asInt() == (int) (float) jsonNode.asInt()))
          .build();
    }
  
    /**
     * Extracts double value from JsonNode if it is within bounds.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<Double> doubleValueExtractor() {
      return ValidatingValueExtractor.<Double>builder()
          .setExtractor(JsonNode::doubleValue)
          .setValidator(
              jsonNode ->
                  jsonNode.isDouble()
  
                      // Either a long number which allows lossless conversion to float
                      || (jsonNode.isIntegralNumber()
                          && jsonNode.canConvertToLong()
                          && jsonNode.asLong() == (long) (double) jsonNode.asInt())
  
                      // Or a decimal number which allows lossless conversion to float
                      || (jsonNode.isFloatingPointNumber()
                          && jsonNode
                              .decimalValue()
                              .equals(BigDecimal.valueOf(jsonNode.doubleValue()))))
          .build();
    }
  
    /**
     * Extracts boolean value from JsonNode if it is within bounds.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<Boolean> booleanValueExtractor() {
      return ValidatingValueExtractor.<Boolean>builder()
          .setExtractor(JsonNode::booleanValue)
          .setValidator(JsonNode::isBoolean)
          .build();
    }
  
    /**
     * Extracts string value from the JsonNode if it is within bounds.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<String> stringValueExtractor() {
      return ValidatingValueExtractor.<String>builder()
          .setExtractor(JsonNode::textValue)
          .setValidator(JsonNode::isTextual)
          .build();
    }
  
    /**
     * Extracts BigDecimal from the JsonNode if it is within bounds.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<BigDecimal> decimalValueExtractor() {
      return ValidatingValueExtractor.<BigDecimal>builder()
          .setExtractor(JsonNode::decimalValue)
          .setValidator(jsonNode -> jsonNode.isNumber())
          .build();
    }
  
    /**
     * Extracts DateTime from the JsonNode (ISO 8601 format string) if it is valid.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<DateTime> datetimeValueExtractor() {
      return ValidatingValueExtractor.<DateTime>builder()
          .setExtractor(jsonNode -> DateTime.parse(jsonNode.textValue()))
          .setValidator(JsonNode::isTextual)
          .build();
    }
  
    /**
     * Extracts LocalDate from the JsonNode (ISO 8601 format string) if it is valid.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<LocalDate> dateValueExtractor() {
      return ValidatingValueExtractor.<LocalDate>builder()
          .setExtractor(jsonNode -> LocalDate.parse(jsonNode.textValue()))
          .setValidator(JsonNode::isTextual)
          .build();
    }
  
    /**
     * Extracts LocalTime from the JsonNode (ISO 8601 format string) if it is valid.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<LocalTime> timeValueExtractor() {
      return ValidatingValueExtractor.<LocalTime>builder()
          .setExtractor(jsonNode -> LocalTime.parse(jsonNode.textValue()))
          .setValidator(JsonNode::isTextual)
          .build();
    }
  
    /**
     * Extracts LocalDateTime from the JsonNode (ISO 8601 format string) if it is valid.
     *
     * <p>Throws {@link UnsupportedRowJsonException} if value is out of bounds.
     */
    static ValueExtractor<LocalDateTime> localDatetimeValueExtractor() {
      return ValidatingValueExtractor.<LocalDateTime>builder()
          .setExtractor(jsonNode -> LocalDateTime.parse(jsonNode.textValue()))
          .setValidator(JsonNode::isTextual)
          .build();
    }
  
    @AutoValue
    public abstract static class ValidatingValueExtractor<W> implements ValueExtractor<W> {
  
      abstract Predicate<JsonNode> validator();
  
      abstract Function<JsonNode, W> extractor();
  
      static <T> Builder<T> builder() {
        return new AutoValue_RowJsonValueExtractors_ValidatingValueExtractor.Builder<>();
      }
  
      @Override
      public W extractValue(JsonNode value) {
        if (!validator().test(value)) {
          throw new UnsupportedRowJsonException(
              "Value \""
                  + value.asText()
                  + "\" "
                  + "is out of range for the type of the field defined in the row schema.");
        }
  
        return extractor().apply(value);
      }
  
      @AutoValue.Builder
      abstract static class Builder<W> {
        abstract Builder<W> setValidator(Predicate<JsonNode> validator);
  
        abstract Builder<W> setExtractor(Function<JsonNode, W> extractor);
  
        abstract ValidatingValueExtractor<W> build();
      }
    }
  }
  