[FLINK-13450][test] Use StrictMath instead of Math

flink-clients/src/main/java/org/apache/flink/client/program/rest/retry/ExponentialWaitStrategy.java

@@ -40,7 +40,7 @@ public ExponentialWaitStrategy(final long initialWait, final long maxWait) {
  @Override
  public long sleepTime(final long attempt) {
  checkArgument(attempt >= 0, "attempt must not be negative (%s)", attempt);
- final long exponentialSleepTime = initialWait * Math.round(Math.pow(2, attempt));
+ final long exponentialSleepTime = initialWait * Math.round(StrictMath.pow(2, attempt));
  return exponentialSleepTime >= 0 && exponentialSleepTime < maxWait ? exponentialSleepTime : maxWait;
  }
 }

flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java

@@ -550,7 +550,7 @@ protected DescribeStreamResult describeStream(String streamName, @Nullable Strin
  }
 
  protected static long fullJitterBackoff(long base, long max, double power, int attempt) {
- long exponentialBackoff = (long) Math.min(max, base * Math.pow(power, attempt));
+ long exponentialBackoff = (long) Math.min(max, base * StrictMath.pow(power, attempt));
  return (long) (seed.nextDouble() * exponentialBackoff); // random jitter between 0 and the exponential backoff
  }
 }

flink-end-to-end-tests/flink-tpch-test/src/main/java/org/apache/flink/table/tpch/TpchResultComparator.java

@@ -115,9 +115,9 @@ private static double round(double x, int m) {
  if (x < 0) {
  throw new IllegalArgumentException("x must be non-negative");
  }
- double y = x + 5 * Math.pow(10, -m - 1);
- double z = y * Math.pow(10, m);
+ double y = x + 5 * StrictMath.pow(10, -m - 1);
+ double z = y * StrictMath.pow(10, m);
  double q = Math.floor(z);
- return q / Math.pow(10, m);
+ return q / StrictMath.pow(10, m);
  }
 }

flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/utils/AvroTestUtils.java

@@ -97,7 +97,7 @@ public static Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> getSp
  // unscaled integer value in big-endian byte order
  .setTypeDecimalBytes(ByteBuffer.wrap(BigDecimal.valueOf(2000, 2).unscaledValue().toByteArray()))
  // array of length n can store at most
- // Math.floor(Math.log10(Math.pow(2, 8 * n - 1) - 1))
+ // Math.floor(Math.log10(StrictMath.pow(2, 8 * n - 1) - 1))
  // base-10 digits of precision
  .setTypeDecimalFixed(new Fixed2(BigDecimal.valueOf(2000, 2).unscaledValue().toByteArray()))
  .build();

flink-java/src/main/java/org/apache/flink/api/java/sampling/BernoulliSampler.java

@@ -113,7 +113,7 @@ private T getNextSampledElement() {
  if (fraction <= THRESHOLD) {
  double rand = random.nextDouble();
  double u = Math.max(rand, EPSILON);
- int gap = (int) (Math.log(u) / Math.log(1 - fraction));
+ int gap = (int) (StrictMath.log(u) / StrictMath.log(1 - fraction));
  int elementCount = 0;
  if (input.hasNext()) {
  T element = input.next();

flink-java/src/main/java/org/apache/flink/api/java/sampling/PoissonSampler.java

@@ -117,7 +117,7 @@ public T next() {
 
  public int poisson_ge1(double p) {
  // sample 'k' from Poisson(p), conditioned to k >= 1.
- double q = Math.pow(Math.E, -p);
+ double q = StrictMath.pow(Math.E, -p);
  // simulate a poisson trial such that k >= 1.
  double t = q + (1 - q) * random.nextDouble();
  int k = 1;
@@ -142,7 +142,7 @@ private void skipGapElements(int num) {
  private void samplingProcess() {
  if (fraction <= THRESHOLD) {
  double u = Math.max(random.nextDouble(), EPSILON);
- int gap = (int) (Math.log(u) / -fraction);
+ int gap = (int) (StrictMath.log(u) / -fraction);
  skipGapElements(gap);
  if (input.hasNext()) {
  currentElement = input.next();

flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/similarity/AdamicAdar.java

@@ -240,7 +240,7 @@ public Tuple3<T, LongValue, FloatValue> map(Vertex<T, LongValue> value)
 
  long degree = value.f1.getValue();
  // when the degree is one the logarithm is zero so avoid dividing by this value
- float inverseLogDegree = (degree == 1) ? 0.0f : 1.0f / (float) Math.log(value.f1.getValue());
+ float inverseLogDegree = (degree == 1) ? 0.0f : 1.0f / (float) StrictMath.log(value.f1.getValue());
  output.f2.setValue(inverseLogDegree);
 
  return output;

flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/similarity/AdamicAdarTest.java

@@ -41,12 +41,12 @@
 public class AdamicAdarTest extends AsmTestBase {
 
  private float[] ilog = {
- 1.0f / (float) Math.log(2),
- 1.0f / (float) Math.log(3),
- 1.0f / (float) Math.log(3),
- 1.0f / (float) Math.log(4),
- 1.0f / (float) Math.log(1),
- 1.0f / (float) Math.log(1)
+ 1.0f / (float) StrictMath.log(2),
+ 1.0f / (float) StrictMath.log(3),
+ 1.0f / (float) StrictMath.log(3),
+ 1.0f / (float) StrictMath.log(4),
+ 1.0f / (float) StrictMath.log(1),
+ 1.0f / (float) StrictMath.log(1)
  };
 
  @Test
@@ -128,7 +128,7 @@ public void testWithCompleteGraph() throws Exception {
  // all vertex pairs are linked
  long expectedCount = CombinatoricsUtils.binomialCoefficient((int) completeGraphVertexCount, 2);
 
- float expectedScore = (completeGraphVertexCount - 2) / (float) Math.log(completeGraphVertexCount - 1);
+ float expectedScore = (completeGraphVertexCount - 2) / (float) StrictMath.log(completeGraphVertexCount - 1);
 
  validate(completeGraph, expectedCount, expectedScore);
  }
@@ -150,7 +150,7 @@ public void testWithStarGraph() throws Exception {
  long expectedCount = CombinatoricsUtils.binomialCoefficient((int) starGraphVertexCount - 1, 2);
 
  // the intersection includes only the center vertex
- float expectedScore = 1 / (float) Math.log(starGraphVertexCount - 1);
+ float expectedScore = 1 / (float) StrictMath.log(starGraphVertexCount - 1);
 
  validate(starGraph, expectedCount, expectedScore);
  }

flink-ml-parent/flink-ml-lib/src/main/java/org/apache/flink/ml/common/linalg/DenseVector.java

@@ -268,9 +268,9 @@ public void normalizeEqual(double p) {
  norm = normL2();
  } else {
  for (int i = 0; i < data.length; i++) {
- norm += Math.pow(Math.abs(data[i]), p);
+ norm += StrictMath.pow(Math.abs(data[i]), p);
  }
- norm = Math.pow(norm, 1 / p);
+ norm = StrictMath.pow(norm, 1 / p);
  }
  for (int i = 0; i < data.length; i++) {
  data[i] /= norm;

flink-ml-parent/flink-ml-lib/src/main/java/org/apache/flink/ml/common/linalg/SparseVector.java

@@ -526,9 +526,9 @@ public void normalizeEqual(double p) {
  norm = normL2();
  } else {
  for (int i = 0; i < indices.length; i++) {
- norm += Math.pow(values[i], p);
+ norm += StrictMath.pow(values[i], p);
  }
- norm = Math.pow(norm, 1 / p);
+ norm = StrictMath.pow(norm, 1 / p);
  }
 
  for (int i = 0; i < indices.length; i++) {

flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java

@@ -1425,7 +1425,7 @@ public MemorySegment nextSegment() {
  * @return The number 
  */
  public static int getNumWriteBehindBuffers(int numBuffers) {
- int numIOBufs = (int) (Math.log(numBuffers) / Math.log(4) - 1.5);
+ int numIOBufs = (int) (StrictMath.log(numBuffers) / StrictMath.log(4) - 1.5);
  return numIOBufs > 6 ? 6 : numIOBufs;
  }
 

flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/UnilateralSortMerger.java

@@ -1575,10 +1575,10 @@ protected final List<ChannelWithBlockCount> mergeChannelList(final List<ChannelW
  // A channel list with length maxFanIn<sup>i</sup> can be merged to maxFanIn files in i-1 rounds where every merge
  // is a full merge with maxFanIn input channels. A partial round includes merges with fewer than maxFanIn
  // inputs. It is most efficient to perform the partial round first.
- final double scale = Math.ceil(Math.log(channelIDs.size()) / Math.log(this.maxFanIn)) - 1;
+ final double scale = Math.ceil(StrictMath.log(channelIDs.size()) / StrictMath.log(this.maxFanIn)) - 1;
 
  final int numStart = channelIDs.size();
- final int numEnd = (int) Math.pow(this.maxFanIn, scale);
+ final int numEnd = (int) StrictMath.pow(this.maxFanIn, scale);
 
  final int numMerges = (int) Math.ceil((numStart - numEnd) / (double) (this.maxFanIn - 1));
 

flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/BloomFilter.java

@@ -67,7 +67,7 @@ public void setBitsLocation(MemorySegment memorySegment, int offset) {
  * @return optimal bits number
  */
  public static int optimalNumOfBits(long inputEntries, double fpp) {
- int numBits = (int) (-inputEntries * Math.log(fpp) / (Math.log(2) * Math.log(2)));
+ int numBits = (int) (-inputEntries * StrictMath.log(fpp) / (StrictMath.log(2) * StrictMath.log(2)));
  return numBits;
  }
 
@@ -81,8 +81,8 @@ public static int optimalNumOfBits(long inputEntries, double fpp) {
  */
  public static double estimateFalsePositiveProbability(long inputEntries, int bitSize) {
  int numFunction = optimalNumOfHashFunctions(inputEntries, bitSize);
- double p = Math.pow(Math.E, -(double) numFunction * inputEntries / bitSize);
- double estimatedFPP = Math.pow(1 - p, numFunction);
+ double p = StrictMath.pow(Math.E, -(double) numFunction * inputEntries / bitSize);
+ double estimatedFPP = StrictMath.pow(1 - p, numFunction);
  return estimatedFPP;
  }
 
@@ -95,7 +95,7 @@ public static double estimateFalsePositiveProbability(long inputEntries, int bit
  * @return hash function number
  */
  static int optimalNumOfHashFunctions(long expectEntries, long bitSize) {
- return Math.max(1, (int) Math.round((double) bitSize / expectEntries * Math.log(2)));
+ return Math.max(1, (int) Math.round((double) bitSize / expectEntries * StrictMath.log(2)));
  }
 
  public void addHash(int hash32) {

flink-runtime/src/test/java/org/apache/flink/runtime/io/disk/iomanager/IOManagerITCase.java

@@ -211,7 +211,7 @@ public void parallelChannelsTest() throws Exception {
 
  private static int skewedSample(Random rnd, int max) {
  double uniform = rnd.nextDouble();
- double var = Math.pow(uniform, 8.0);
+ double var = StrictMath.pow(uniform, 8.0);
  double pareto = 0.2 / var;
 
  int val = (int) pareto;

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/cost/FlinkCost.scala

@@ -252,7 +252,7 @@ class FlinkCost(
  if (n == 0) {
  return 1.0
  }
- Math.pow(d, 1 / n)
+ StrictMath.pow(d, 1 / n)
  }
 
  override def hashCode(): Int = {

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecSort.scala

@@ -84,7 +84,7 @@ class BatchExecSort(
  }
  val numOfSortKeys = sortCollation.getFieldCollations.size()
  val cpuCost = FlinkCost.COMPARE_CPU_COST * numOfSortKeys *
- rowCount * Math.max(Math.log(rowCount), 1.0)
+ rowCount * Math.max(StrictMath.log(rowCount), 1.0)
  val memCost = FlinkRelMdUtil.computeSortMemory(mq, getInput)
  val costFactory = planner.getCostFactory.asInstanceOf[FlinkCostFactory]
  costFactory.makeCost(rowCount, cpuCost, 0, 0, memCost)

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecSortLimit.scala

@@ -102,7 +102,7 @@ class BatchExecSortLimit(
  val inputRowCnt = mq.getRowCount(getInput())
  val heapLen = Math.min(inputRowCnt, limitEnd)
  val numOfSort = sortCollation.getFieldCollations.size()
- val cpuCost = FlinkCost.COMPARE_CPU_COST * numOfSort * inputRowCnt * Math.log(heapLen)
+ val cpuCost = FlinkCost.COMPARE_CPU_COST * numOfSort * inputRowCnt * StrictMath.log(heapLen)
  // assume memory is big enough to simplify the estimation.
  val memCost = heapLen * mq.getAverageRowSize(this)
  val rowCount = mq.getRowCount(this)

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecSortMergeJoin.scala

@@ -108,14 +108,16 @@ class BatchExecSortMergeJoin(
  leftRowCnt
  } else {
  // sort cost
- FlinkCost.COMPARE_CPU_COST * numOfSort * leftRowCnt * Math.max(Math.log(leftRowCnt), 1.0)
+ FlinkCost.COMPARE_CPU_COST * numOfSort * leftRowCnt * Math.max(
+ StrictMath.log(leftRowCnt), 1.0)
  }
  val rightSortCpuCost: Double = if (rightSorted) {
  // cost of writing rhs data to buffer
  rightRowCnt
  } else {
  // sort cost
- FlinkCost.COMPARE_CPU_COST * numOfSort * rightRowCnt * Math.max(Math.log(rightRowCnt), 1.0)
+ FlinkCost.COMPARE_CPU_COST * numOfSort * rightRowCnt * Math.max(
+ StrictMath.log(rightRowCnt), 1.0)
  }
  // cost of evaluating each join condition
  val joinConditionCpuCost = FlinkCost.COMPARE_CPU_COST * (leftRowCnt + rightRowCnt)

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelMdUtil.scala

@@ -130,7 +130,7 @@ object FlinkRelMdUtil {
  distinctRowCount: JDouble,
  selectivity: JDouble): JDouble = {
  val ndv = Math.min(distinctRowCount, rowCount)
- Math.max((1 - Math.pow(1 - selectivity, rowCount / ndv)) * ndv, 1.0)
+ Math.max((1 - StrictMath.pow(1 - selectivity, rowCount / ndv)) * ndv, 1.0)
  }
 
  /**

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/cost/FlinkCostTest.scala

@@ -180,8 +180,8 @@ class FlinkCostTest {
 
  val cost1 = FlinkCost.FACTORY.makeCost(100.0, 1000.0, 200.0, 500.0, 600.0)
  val cost2 = FlinkCost.FACTORY.makeCost(50.0, 100.0, 40.0, 25.0, 200.0)
- assertEquals(Math.pow(6000.0, 1 / 5.0), cost1.divideBy(cost2), 1e-5)
- assertEquals(Math.pow(1 / 6000.0, 1 / 5.0), cost2.divideBy(cost1), 1e-5)
+ assertEquals(StrictMath.pow(6000.0, 1 / 5.0), cost1.divideBy(cost2), 1e-5)
+ assertEquals(StrictMath.pow(1 / 6000.0, 1 / 5.0), cost2.divideBy(cost1), 1e-5)
  }
 
 }

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/DecimalITCase.scala

@@ -490,7 +490,8 @@ class DecimalITCase extends BatchTestBase {
  s1r(d"3.14"),
  "select log10(f0), ln(f0), log(f0), log2(f0) from Table1",
  Seq(DOUBLE, DOUBLE, DOUBLE, DOUBLE),
- s1r(log10(3.14), Math.log(3.14), Math.log(3.14), Math.log(3.14)/Math.log(2.0)))
+ s1r(log10(3.14), StrictMath.log(3.14), StrictMath.log(3.14),
+ StrictMath.log(3.14)/StrictMath.log(2.0)))
 
  checkQuery1(
  Seq(DECIMAL(10, 2), DOUBLE),

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/cost/DataSetCost.scala

@@ -109,7 +109,7 @@ class DataSetCost(val rowCount: Double, val cpu: Double, val io: Double) extends
  if (n == 0) {
  return 1.0
  }
- Math.pow(d, 1 / n)
+ StrictMath.pow(d, 1 / n)
  }
 
  def plus(other: RelOptCost): RelOptCost = {

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/functions/ScalarFunctions.scala

@@ -39,7 +39,7 @@ class ScalarFunctions {}
 object ScalarFunctions {
 
  def power(a: Double, b: JBigDecimal): Double = {
- Math.pow(a, b.doubleValue())
+ StrictMath.pow(a, b.doubleValue())
  }
 
  /**
@@ -107,7 +107,7 @@ object ScalarFunctions {
  if (x <= 0.0) {
  throw new IllegalArgumentException(s"x of 'log(x)' must be > 0, but x = $x")
  } else {
- Math.log(x)
+ StrictMath.log(x)
  }
  }
 
@@ -135,7 +135,7 @@ object ScalarFunctions {
  if (base <= 1.0) {
  throw new IllegalArgumentException(s"base of 'log(base, x)' must be > 1, but base = $base")
  } else {
- Math.log(x) / Math.log(base)
+ StrictMath.log(x) / StrictMath.log(base)
  }
  }
 
@@ -146,7 +146,7 @@ object ScalarFunctions {
  if (x <= 0.0) {
  throw new IllegalArgumentException(s"x of 'log2(x)' must be > 0, but x = $x")
  } else {
- Math.log(x) / Math.log(2)
+ StrictMath.log(x) / StrictMath.log(2)
  }
  }
 

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java

@@ -85,15 +85,15 @@ public static double exp(Decimal d) {
  }
 
  public static double power(double base, Decimal exponent) {
- return Math.pow(base, exponent.doubleValue());
+ return StrictMath.pow(base, exponent.doubleValue());
  }
 
  public static double power(Decimal base, Decimal exponent) {
- return Math.pow(base.doubleValue(), exponent.doubleValue());
+ return StrictMath.pow(base.doubleValue(), exponent.doubleValue());
  }
 
  public static double power(Decimal base, double exponent) {
- return Math.pow(base.doubleValue(), exponent);
+ return StrictMath.pow(base.doubleValue(), exponent);
  }
 
  public static double cosh(Decimal x) {
@@ -169,18 +169,18 @@ public static Decimal ceil(Decimal a) {
  * Returns the natural logarithm of "x".
  */
  public static double log(double x) {
- return Math.log(x);
+ return StrictMath.log(x);
  }
 
  public static double log(Decimal x) {
- return Math.log(x.doubleValue());
+ return StrictMath.log(x.doubleValue());
  }
 
  /**
  * Returns the logarithm of "x" with base "base".
  */
  public static double log(double base, double x) {
- return Math.log(x) / Math.log(base);
+ return StrictMath.log(x) / StrictMath.log(base);
  }
 
  public static double log(double base, Decimal x) {
@@ -199,7 +199,7 @@ public static double log(Decimal base, Decimal x) {
  * Returns the logarithm of "a" with base 2.
  */
  public static double log2(double x) {
- return Math.log(x) / Math.log(2);
+ return StrictMath.log(x) / StrictMath.log(2);
  }
 
  public static double log2(Decimal x) {

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryHashTable.java

@@ -334,7 +334,7 @@ public RowIterator<BinaryRow> getBuildSideIterator() {
  */
  @VisibleForTesting
  static int getNumWriteBehindBuffers(int numBuffers) {
- int numIOBufs = (int) (Math.log(numBuffers) / Math.log(4) - 1.5);
+ int numIOBufs = (int) (StrictMath.log(numBuffers) / StrictMath.log(4) - 1.5);
  return numIOBufs > 6 ? 6 : numIOBufs;
  }
 

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/HashTableBloomFilter.java

@@ -83,6 +83,6 @@ static int optimalSegmentNumber(long maxNum, int segSize, double fpp) {
  }
 
  private static int optimalNumOfBits(long maxNumEntries, double fpp) {
- return (int) (-maxNumEntries * Math.log(fpp) / (Math.log(2) * Math.log(2)));
+ return (int) (-maxNumEntries * StrictMath.log(fpp) / (StrictMath.log(2) * StrictMath.log(2)));
  }
 }

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/sort/AbstractBinaryExternalMerger.java

@@ -120,10 +120,10 @@ public List<ChannelWithMeta> mergeChannelList(List<ChannelWithMeta> channelIDs)
  // A channel list with length maxFanIn<sup>i</sup> can be merged to maxFanIn files in i-1 rounds where every merge
  // is a full merge with maxFanIn input channels. A partial round includes merges with fewer than maxFanIn
  // inputs. It is most efficient to perform the partial round first.
- final double scale = Math.ceil(Math.log(channelIDs.size()) / Math.log(maxFanIn)) - 1;
+ final double scale = Math.ceil(StrictMath.log(channelIDs.size()) / StrictMath.log(maxFanIn)) - 1;
 
  final int numStart = channelIDs.size();
- final int numEnd = (int) Math.pow(maxFanIn, scale);
+ final int numEnd = (int) StrictMath.pow(maxFanIn, scale);
 
  final int numMerges = (int) Math.ceil((numStart - numEnd) / (double) (maxFanIn - 1));