<commit_before>/*
 * 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.flink.table.planner.plan.metadata

import org.apache.flink.table.api.TableConfig
import org.apache.flink.table.catalog.FunctionCatalog
import org.apache.flink.table.module.ModuleManager
import org.apache.flink.table.plan.stats.{ColumnStats, TableStats}
import org.apache.flink.table.planner.calcite.{FlinkContext, FlinkContextImpl, FlinkTypeFactory, FlinkTypeSystem}
import org.apache.flink.table.planner.plan.schema._
import org.apache.flink.table.planner.plan.stats.FlinkStatistic
import org.apache.flink.table.planner.{JDouble, JLong}
import org.apache.flink.table.utils.CatalogManagerMocks
import org.apache.flink.util.Preconditions

import org.apache.calcite.plan.{AbstractRelOptPlanner, RelOptCluster}
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.TableScan
import org.apache.calcite.rel.metadata.{JaninoRelMetadataProvider, RelMetadataQueryBase}
import org.apache.calcite.rex.{RexBuilder, RexInputRef, RexLiteral, RexNode}
import org.apache.calcite.sql.SqlOperator
import org.apache.calcite.sql.`type`.SqlTypeName._
import org.apache.calcite.sql.fun.SqlStdOperatorTable
import org.apache.calcite.sql.fun.SqlStdOperatorTable._
import org.apache.calcite.util.{DateString, TimeString, TimestampString}
import org.junit.Assert._
import org.junit.runner.RunWith
import org.junit.{Before, BeforeClass, Test}
import org.powermock.api.mockito.PowerMockito._
import org.powermock.core.classloader.annotations.PrepareForTest
import org.powermock.modules.junit4.PowerMockRunner

import java.math.BigDecimal
import java.sql.{Date, Time, Timestamp}

import scala.collection.JavaConverters._

/**
  * Tests for [[SelectivityEstimator]].
  *
  * We use PowerMockito instead of Mockito here, because [[TableScan#getRowType]] is a final method.
  */
@RunWith(classOf[PowerMockRunner])
@PrepareForTest(Array(classOf[TableScan]))
class SelectivityEstimatorTest {
  private val allFieldNames = Seq("name", "amount", "price", "flag", "partition",
    "date_col", "time_col", "timestamp_col")
  private val allFieldTypes = Seq(VARCHAR, INTEGER, DOUBLE, BOOLEAN, VARCHAR,
    DATE, TIME, TIMESTAMP)
  val (name_idx, amount_idx, price_idx, flag_idx, partition_idx,
  date_idx, time_idx, timestamp_idx) = (0, 1, 2, 3, 4, 5, 6, 7)

  val typeFactory: FlinkTypeFactory = new FlinkTypeFactory(new FlinkTypeSystem())
  var rexBuilder = new RexBuilder(typeFactory)
  val relDataType: RelDataType = typeFactory.createStructType(
    allFieldTypes.map(typeFactory.createSqlType).asJava,
    allFieldNames.asJava)

  val mq: FlinkRelMetadataQuery = FlinkRelMetadataQuery.instance()
  var scan: TableScan = _

  @Before
  def setup(): Unit = {
    scan = mockScan()
  }

  private def mockScan(
      statistic: FlinkStatistic = FlinkStatistic.UNKNOWN,
      isFilterPushedDown: Boolean = false,
      tableConfig: TableConfig = TableConfig.getDefault): TableScan = {
    val tableScan = mock(classOf[TableScan])
    val cluster = mock(classOf[RelOptCluster])
    val planner = mock(classOf[AbstractRelOptPlanner])
    val catalogManager = CatalogManagerMocks.createEmptyCatalogManager()
    val moduleManager = mock(classOf[ModuleManager])
    val functionCatalog = new FunctionCatalog(tableConfig, catalogManager, moduleManager)
    val context: FlinkContext = new FlinkContextImpl(
      tableConfig, functionCatalog, catalogManager, null)
    when(tableScan, "getCluster").thenReturn(cluster)
    when(cluster, "getRexBuilder").thenReturn(rexBuilder)
    when(cluster, "getPlanner").thenReturn(planner)
    when(planner, "getContext").thenReturn(context)
    when(tableScan, "getRowType").thenReturn(relDataType)
    val sourceTable = mock(classOf[TableSourceTable[_]])
    when(sourceTable, "unwrap", classOf[TableSourceTable[_]]).thenReturn(sourceTable)
    when(sourceTable, "getStatistic").thenReturn(statistic)
    when(sourceTable, "getRowType").thenReturn(relDataType)
    when(tableScan, "getTable").thenReturn(sourceTable)
    val rowCount: JDouble = if (statistic != null && statistic.getRowCount != null) {
      statistic.getRowCount
    } else {
      100D
    }
    when(tableScan, "estimateRowCount", mq).thenReturn(rowCount)
    tableScan
  }

  private def createNumericLiteral(num: Long): RexLiteral = {
    rexBuilder.makeExactLiteral(BigDecimal.valueOf(num))
  }

  private def createNumericLiteral(num: Double): RexLiteral = {
    rexBuilder.makeExactLiteral(BigDecimal.valueOf(num))
  }

  private def createBooleanLiteral(b: Boolean): RexLiteral = {
    rexBuilder.makeLiteral(b)
  }
<commit_msg>[FLINK-16476] [table-planner-blink] Remove PowerMockito to avoid LinkageError in SelectivityEstimatorTest<commit_after>/*
 * 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.flink.table.planner.plan.metadata

import org.apache.flink.table.api.TableConfig
import org.apache.flink.table.catalog.FunctionCatalog
import org.apache.flink.table.module.ModuleManager
import org.apache.flink.table.plan.stats.{ColumnStats, TableStats}
import org.apache.flink.table.planner.calcite.{FlinkTypeFactory, FlinkTypeSystem}
import org.apache.flink.table.planner.delegation.PlannerContext
import org.apache.flink.table.planner.plan.`trait`.FlinkRelDistributionTraitDef
import org.apache.flink.table.planner.plan.stats.FlinkStatistic
import org.apache.flink.table.planner.{JDouble, JLong}
import org.apache.flink.table.utils.CatalogManagerMocks
import org.apache.flink.util.Preconditions

import org.apache.calcite.jdbc.CalciteSchema
import org.apache.calcite.plan.ConventionTraitDef
import org.apache.calcite.rel.RelCollationTraitDef
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.TableScan
import org.apache.calcite.rel.metadata.{JaninoRelMetadataProvider, RelMetadataQueryBase}
import org.apache.calcite.rex.{RexBuilder, RexInputRef, RexLiteral, RexNode}
import org.apache.calcite.sql.SqlOperator
import org.apache.calcite.sql.`type`.SqlTypeName._
import org.apache.calcite.sql.fun.SqlStdOperatorTable
import org.apache.calcite.sql.fun.SqlStdOperatorTable._
import org.apache.calcite.util.{DateString, TimeString, TimestampString}
import org.junit.Assert._
import org.junit.{Before, BeforeClass, Test}

import java.math.BigDecimal
import java.sql.{Date, Time, Timestamp}
import java.util

import scala.collection.JavaConverters._

/**
  * Tests for [[SelectivityEstimator]].
  */
class SelectivityEstimatorTest {
  private val allFieldNames = Seq("name", "amount", "price", "flag", "partition",
    "date_col", "time_col", "timestamp_col")
  private val allFieldTypes = Seq(VARCHAR, INTEGER, DOUBLE, BOOLEAN, VARCHAR,
    DATE, TIME, TIMESTAMP)
  val (name_idx, amount_idx, price_idx, flag_idx, partition_idx,
  date_idx, time_idx, timestamp_idx) = (0, 1, 2, 3, 4, 5, 6, 7)

  val typeFactory: FlinkTypeFactory = new FlinkTypeFactory(new FlinkTypeSystem())
  var rexBuilder = new RexBuilder(typeFactory)
  val relDataType: RelDataType = typeFactory.createStructType(
    allFieldTypes.map(typeFactory.createSqlType).asJava,
    allFieldNames.asJava)

  val mq: FlinkRelMetadataQuery = FlinkRelMetadataQuery.instance()
  var scan: TableScan = _

  @Before
  def setup(): Unit = {
    scan = mockScan()
  }

  private def mockScan(
      statistic: FlinkStatistic = FlinkStatistic.UNKNOWN,
      tableConfig: TableConfig = TableConfig.getDefault): TableScan = {
    val catalogManager = CatalogManagerMocks.createEmptyCatalogManager()
    val rootSchema = CalciteSchema.createRootSchema(true, false).plus()
    val table = new MockMetaTable(relDataType, statistic)
    rootSchema.add("test", table)
    val plannerContext: PlannerContext =
      new PlannerContext(
        tableConfig,
        new FunctionCatalog(tableConfig, catalogManager, new ModuleManager),
        catalogManager,
        CalciteSchema.from(rootSchema),
        util.Arrays.asList(
          ConventionTraitDef.INSTANCE,
          FlinkRelDistributionTraitDef.INSTANCE,
          RelCollationTraitDef.INSTANCE
        )
      )

    val relBuilder = plannerContext.createRelBuilder("default_catalog", "default_database")
    relBuilder.clear()
    relBuilder.scan(util.Arrays.asList("test")).build().asInstanceOf[TableScan]
  }

  private def createNumericLiteral(num: Long): RexLiteral = {
    rexBuilder.makeExactLiteral(BigDecimal.valueOf(num))
  }

  private def createNumericLiteral(num: Double): RexLiteral = {
    rexBuilder.makeExactLiteral(BigDecimal.valueOf(num))
  }

  private def createBooleanLiteral(b: Boolean): RexLiteral = {
    rexBuilder.makeLiteral(b)
  }

{'fraction_non_alphanumeric': 0.09923890481367995, 'fraction_numerical': 0.0032618365128002373, 'mean_word_length': 4.997628927089508, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '540601', 'n_tokens_mistral': 3069, 'n_tokens_neox': 3024, 'n_words': 707}