/*
 * 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.plan.metadata

import org.apache.flink.table.calcite.FlinkRelBuilder.NamedWindowProperty
import org.apache.flink.table.plan.FlinkJoinRelType
import org.apache.flink.table.plan.nodes.calcite.{Expand, LogicalWindowAggregate, Rank}
import org.apache.flink.table.plan.nodes.logical._
import org.apache.flink.table.plan.nodes.physical.batch._
import org.apache.flink.table.plan.nodes.physical.stream._
import org.apache.flink.table.plan.schema.FlinkRelOptTable
import org.apache.flink.table.plan.util.{FlinkRelMdUtil, TemporalJoinUtil}
import org.apache.flink.table.sources.{IndexKey, TableSource}

import com.google.common.collect.ImmutableSet
import org.apache.calcite.plan.RelOptTable
import org.apache.calcite.plan.volcano.RelSubset
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.{JoinRelType, _}
import org.apache.calcite.rel.logical.LogicalSnapshot
import org.apache.calcite.rel.metadata._
import org.apache.calcite.rel.{RelNode, SingleRel}
import org.apache.calcite.rex.{RexCall, RexInputRef, RexNode}
import org.apache.calcite.sql.SqlKind
import org.apache.calcite.sql.fun.SqlStdOperatorTable
import org.apache.calcite.util.{Bug, BuiltInMethod, ImmutableBitSet, Util}

import java.lang.{Boolean => JBool}
import java.util
import java.util.{ArrayList => JArrayList, List => JList}

import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
import scala.collection.mutable

class FlinkRelMdUniqueKeys private extends MetadataHandler[BuiltInMetadata.UniqueKeys] {

  def getDef: MetadataDef[BuiltInMetadata.UniqueKeys] = BuiltInMetadata.UniqueKeys.DEF

  private def getTableUniqueKeys(
      tableSource: TableSource,
      relOptTable: RelOptTable): util.Set[ImmutableBitSet] = {
    if (tableSource != null) {
      val indexes = TemporalJoinUtil.getTableIndexKeys(tableSource)
      if (indexes.nonEmpty) {
        return indexKeysToUniqueBitSet(indexes)
      }
    }
    relOptTable match {
      case flinkRelOptTable: FlinkRelOptTable => flinkRelOptTable.uniqueKeysSet.orNull
      case _ => null
    }
  }

  def getUniqueKeys(
      rel: TableScan,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    getTableUniqueKeys(null, rel.getTable)
  }

  def getUniqueKeys(
      snapshot: LogicalSnapshot,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    mq.getUniqueKeys(snapshot.getInput, ignoreNulls)
  }

  def getUniqueKeys(
      rel: FlinkLogicalTableSourceScan,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    getTableUniqueKeys(rel.tableSource, rel.getTable)
  }

  def getUniqueKeys(
      rel: Expand,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    // mapping input column index to output index for non-null value columns
    val mapInputToOutput = new util.HashMap[Int, Int]()
    (0 until rel.getRowType.getFieldCount).filter(_ != rel.expandIdIndex).foreach { column =>
      val inputRefs = FlinkRelMdUtil.getInputRefIndices(column, rel)
      // expand columns corresponding to a given index should be same input ref.
      if (inputRefs.size() == 1 && inputRefs.head >= 0) {
        mapInputToOutput.put(inputRefs.head, column)
      }
    }
    if (mapInputToOutput.isEmpty) {
      return null
    }

    val inputUniqueKeys = mq.getUniqueKeys(rel.getInput, ignoreNulls)
    if (inputUniqueKeys == null || inputUniqueKeys.isEmpty) {
      return inputUniqueKeys
    }

    // values of expand_is are unique in rows expanded from a row,
    // and a input unique key combined with expand_id are also unique
    val outputUniqueKeys = new util.HashSet[ImmutableBitSet]()
    inputUniqueKeys.foreach { uniqueKey =>
      val outputUniqueKeyBuilder = ImmutableBitSet.builder()
      // a input unique key can be output only its values are all in `mapInputToOutput`
      val canOutput = uniqueKey.toList.forall { key =>
        if (mapInputToOutput.contains(key)) {
          outputUniqueKeyBuilder.set(mapInputToOutput.get(key))
          true
        } else {
          false
        }
      }
      if (canOutput) {
        // unique key from input combined with expand id are unique
        outputUniqueKeyBuilder.set(rel.expandIdIndex)
        outputUniqueKeys.add(outputUniqueKeyBuilder.build())
      }
    }
    if (outputUniqueKeys.isEmpty) null else outputUniqueKeys
  }

  def getUniqueKeys(
      rel: Filter,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = mq.getUniqueKeys(rel.getInput, ignoreNulls)

  def getUniqueKeys(
      rel: Sort,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = mq.getUniqueKeys(rel.getInput, ignoreNulls)

  def getUniqueKeys(
      rel: Correlate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = null

  def getUniqueKeys(
      rel: BatchExecCorrelate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = null

  def getUniqueKeys(
      rel: Project,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] =
    getProjectUniqueKeys(rel.getProjects, rel.getInput, mq, ignoreNulls)


  private def getProjectUniqueKeys(
      projects: JList[RexNode],
      input: RelNode,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    // LogicalProject maps a set of rows to a different set;
    // Without knowledge of the mapping function(whether it
    // preserves uniqueness), it is only safe to derive uniqueness
    // info from the child of a project when the mapping is f(a) => a.
    //
    // Further more, the unique bitset coming from the child needs
    val projUniqueKeySet = mutable.HashSet[ImmutableBitSet]()
    val mapInToOutPos = mutable.HashMap[Int, mutable.ArrayBuffer[Int]]()

    def appendMapInToOutPos(inIndex: Int, outIndex: Int): Unit = {
      if (mapInToOutPos.contains(inIndex)) {
        mapInToOutPos(inIndex) += outIndex
      } else {
        val arrayBuffer = mutable.ArrayBuffer[Int]()
        arrayBuffer += outIndex
        mapInToOutPos += (inIndex -> arrayBuffer)
      }
    }
    // Build an input to output position map.
    projects.zipWithIndex.foreach {
      case (projExpr, i) =>
        projExpr match {
          case ref: RexInputRef => appendMapInToOutPos(ref.getIndex, i)
          case a: RexCall if ignoreNulls && a.getOperator.equals(SqlStdOperatorTable.CAST) =>
            val castOperand = a.getOperands.get(0)
            castOperand match {
              case castRef: RexInputRef =>
                val typeFactory = input.getCluster.getTypeFactory
                val castType = typeFactory.createTypeWithNullability(projExpr.getType, true)
                val origType = typeFactory.createTypeWithNullability(castOperand.getType, true)
                if (castType == origType) {
                  appendMapInToOutPos(castRef.getIndex, i)
                }
              case _ => // ignore
            }
          //rename
          case a: RexCall if a.getKind.equals(SqlKind.AS) &&
            a.getOperands.get(0).isInstanceOf[RexInputRef] =>
            appendMapInToOutPos(a.getOperands.get(0).asInstanceOf[RexInputRef].getIndex, i)
          case _ => // ignore
        }
    }
    if (mapInToOutPos.isEmpty) {
      // if there's no RexInputRef in the projected expressions
      // return empty set.
      return projUniqueKeySet
    }

    val childUniqueKeySet = mq.getUniqueKeys(input, ignoreNulls)
    if (childUniqueKeySet != null) {
      // Now add to the projUniqueKeySet the child keys that are fully
      // projected.
      childUniqueKeySet.foreach { colMask =>
        val filerInToOutPos = mapInToOutPos.filter { entry =>
          colMask.asList().contains(entry._1)
        }
        val keys = filerInToOutPos.keys
        if (colMask.forall(keys.contains(_))) {
          val total = filerInToOutPos.map(_._2.size).product
          for (i <- 0 to total) {
            val tmpMask = ImmutableBitSet.builder()
            filerInToOutPos.foreach { entry =>
              tmpMask.set(entry._2.get(i % entry._2.size))
            }
            projUniqueKeySet += tmpMask.build()
          }
        }
      }
    }
    projUniqueKeySet
  }

  def getUniqueKeys(
      join: Join,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    getJoinUniqueKeys(
      join.analyzeCondition(), join.getJoinType, join.getLeft, join.getRight, mq, ignoreNulls)
  }

  private def getJoinUniqueKeys(
      joinInfo: JoinInfo,
      joinRelType: JoinRelType,
      left: RelNode,
      right: RelNode,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    val leftUniqueKeys = mq.getUniqueKeys(left, ignoreNulls)
    val rightUniqueKeys = mq.getUniqueKeys(right, ignoreNulls)
    getJoinUniqueKeys(
      joinInfo, joinRelType, left.getRowType, leftUniqueKeys, rightUniqueKeys,
      mq.areColumnsUnique(left, joinInfo.leftSet, ignoreNulls),
      mq.areColumnsUnique(right, joinInfo.rightSet, ignoreNulls),
      mq)
  }

  private def getJoinUniqueKeys(
      joinInfo: JoinInfo,
      joinRelType: JoinRelType,
      leftType: RelDataType,
      leftUniqueKeys: util.Set[ImmutableBitSet],
      rightUniqueKeys: util.Set[ImmutableBitSet],
      isleftUnique: JBool,
      isRightUnique: JBool,
      mq: RelMetadataQuery): util.Set[ImmutableBitSet] = {

    // first add the different combinations of concatenated unique keys
    // from the left and the right, adjusting the right hand side keys to
    // reflect the addition of the left hand side
    //
    // NOTE zfong 12/18/06 - If the number of tables in a join is large,
    // the number of combinations of unique key sets will explode.  If
    // that is undesirable, use RelMetadataQuery.areColumnsUnique() as
    // an alternative way of getting unique key information.
    val retSet = new util.HashSet[ImmutableBitSet]
    val nFieldsOnLeft = leftType.getFieldCount
    val rightSet = if (rightUniqueKeys != null) {
      val res = new util.HashSet[ImmutableBitSet]
      rightUniqueKeys.foreach { colMask =>
        val tmpMask = ImmutableBitSet.builder
        colMask.foreach(bit => tmpMask.set(bit + nFieldsOnLeft))
        res.add(tmpMask.build())
      }
      if (leftUniqueKeys != null) {
        res.foreach { colMaskRight =>
          leftUniqueKeys.foreach(colMaskLeft => retSet.add(colMaskLeft.union(colMaskRight)))
        }
      }
      res
    } else {
      null
    }

    // determine if either or both the LHS and RHS are unique on the
    // equijoin columns
    val leftUnique = isleftUnique
    val rightUnique = isRightUnique

    // if the right hand side is unique on its equijoin columns, then we can
    // add the unique keys from left if the left hand side is not null
    // generating
    if (rightUnique != null
      && rightUnique
      && (leftUniqueKeys != null)
      && !joinRelType.generatesNullsOnLeft) {
      retSet.addAll(leftUniqueKeys)
    }

    // same as above except left and right are reversed
    if (leftUnique != null
      && leftUnique
      && (rightSet != null)
      && !joinRelType.generatesNullsOnRight) {
      retSet.addAll(rightSet)
    }
    retSet
  }

  def getUniqueKeys(
      rel: SemiJoin,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] =
  // only return the unique keys from the LHS since a semijoin only
  // returns the LHS
    mq.getUniqueKeys(rel.getLeft, ignoreNulls)

  def getUniqueKeys(
      rel: Aggregate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] =
  // group by keys form a unique key
    ImmutableSet.of(ImmutableBitSet.range(rel.getGroupCount))

  def getUniqueKeys(
      rel: BatchExecGroupAggregateBase,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    if (rel.isFinal) {
      // group by keys form a unique key
      ImmutableSet.of(ImmutableBitSet.of(rel.getGrouping: _*))
    } else {
      null
    }
  }

  def getUniqueKeys(
      rel: SetOp,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    if (!rel.all) {
      ImmutableSet.of(ImmutableBitSet.range(rel.getRowType.getFieldCount))
    } else {
      ImmutableSet.of()
    }
  }

  def getUniqueKeys(
      calc: Calc,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    val input = calc.getInput
    val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
    getProjectUniqueKeys(projects, input, mq, ignoreNulls)
  }

  def getUniqueKeys(
      rel: StreamExecCorrelate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = null

  def getUniqueKeys(
      rel: Exchange,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = mq.getUniqueKeys(rel.getInput, ignoreNulls)

  def getUniqueKeys(
      rel: StreamExecGroupAggregate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    // group by keys form a unique key
    toImmutableSet(rel.getGroupings.indices.toArray)
  }

  def getUniqueKeys(
      rel: StreamExecGlobalGroupAggregate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    ImmutableSet.of(ImmutableBitSet.of(rel.getGroupings.indices.toArray: _*))
  }

  def getUniqueKeys(
      rel: StreamExecLocalGroupAggregate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = null

  private def getGroupWindowUniqueKeys(
      fieldCount: Int,
      namedProperties: Seq[NamedWindowProperty],
      grouping: Array[Int],
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    if (namedProperties.nonEmpty) {
      val begin = fieldCount - namedProperties.size
      val end = fieldCount - 1
      //namedProperties's indexes is at the end of output record
      val keys = ImmutableBitSet.of(grouping: _*)
      (begin to end).map {
        i => keys.union(ImmutableBitSet.of(i))
      }.toSet[ImmutableBitSet]
    } else {
      null
    }
  }

  def getUniqueKeys(
      rel: StreamExecGroupWindowAggregate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    getGroupWindowUniqueKeys(
      rel.getRowType.getFieldCount, rel.getWindowProperties, rel.getGroupings, mq, ignoreNulls)
  }

  def getUniqueKeys(
      rel: FlinkLogicalWindowAggregate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    getGroupWindowUniqueKeys(
      rel.getRowType.getFieldCount,
      rel.getNamedProperties,
      rel.getGroupSet.toArray,
      mq,
      ignoreNulls)
  }

  def getUniqueKeys(
      rel: LogicalWindowAggregate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    getGroupWindowUniqueKeys(
      rel.getRowType.getFieldCount,
      rel.getNamedProperties,
      rel.getGroupSet.toArray,
      mq,
      ignoreNulls)
  }

  def getUniqueKeys(
      rel: BatchExecWindowAggregateBase,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    if (rel.isFinal) {
      getGroupWindowUniqueKeys(
        rel.getRowType.getFieldCount,
        rel.getNamedProperties,
        rel.getGrouping,
        mq,
        ignoreNulls)
    } else {
      null
    }
  }

  private def getTemporalTableJoinUniqueKeys(
      joinInfo: JoinInfo,
      joinRelType: JoinRelType,
      left: RelNode,
      joinedIndex: Option[IndexKey],
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    val leftUniqueKeys = mq.getUniqueKeys(left, ignoreNulls)
    val leftType = left.getRowType
    val rightUniqueKeys = indexKeyToUniqueBitSet(joinedIndex)
    getJoinUniqueKeys(
      joinInfo, joinRelType, leftType, leftUniqueKeys, rightUniqueKeys,
      mq.areColumnsUnique(left, joinInfo.leftSet, ignoreNulls),
      null != rightUniqueKeys && rightUniqueKeys.exists { uniqueKey =>
        uniqueKey.forall(key => joinInfo.rightKeys.contains(key))
      },
      mq)
  }

  def getUniqueKeys(
      join: StreamExecTemporalTableJoin,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    getTemporalTableJoinUniqueKeys(
      join.joinInfo,
      join.joinType,
      join.getInput,
      join.joinedIndex,
      mq,
      ignoreNulls)
  }

  def getUniqueKeys(
      join: BatchExecTemporalTableJoin,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    getTemporalTableJoinUniqueKeys(
      join.joinInfo,
      join.joinType,
      join.getInput,
      join.joinedIndex,
      mq,
      ignoreNulls)
  }

  def getUniqueKeys(
      rel: FlinkLogicalLastRow,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    mq.getUniqueKeys(rel.getInput, ignoreNulls)
    ImmutableSet.of(ImmutableBitSet.of(rel.getUniqueKeys.map(Integer.valueOf(_)).toIterable.asJava))
  }

  def getUniqueKeys(
      rel: StreamExecLastRow,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    ImmutableSet.of(ImmutableBitSet.of(rel.getUniqueKeys.map(Integer.valueOf(_)).toIterable.asJava))
  }

  def getUniqueKeys(
      rel: Rank,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    val inputUniqueKeys = mq.getUniqueKeys(rel.getInput, ignoreNulls)
    val rankFunColumnIndex = FlinkRelMdUtil.getRankFunColumnIndex(rel)
    if (rankFunColumnIndex < 0) {
      inputUniqueKeys
    } else {
      val retSet = new util.HashSet[ImmutableBitSet]
      rel.rankFunction.kind match {
        case SqlKind.ROW_NUMBER =>
          retSet.add(rel.partitionKey.union(ImmutableBitSet.of(rankFunColumnIndex)))
        case _ => // do nothing
      }
      if (inputUniqueKeys != null && inputUniqueKeys.nonEmpty) {
        inputUniqueKeys.foreach {
          uniqueKey => retSet.add(uniqueKey)
        }
      }
      retSet
    }
  }

  def getUniqueKeys(
      rel: StreamExecWindowJoin,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.joinCondition)
    getJoinUniqueKeys(joinInfo, rel.joinType, rel.getLeft, rel.getRight, mq, ignoreNulls)
  }

  def getUniqueKeys(
      rel: StreamExecJoin,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    rel.joinType match {
      case FlinkJoinRelType.ANTI | FlinkJoinRelType.SEMI =>
        mq.getUniqueKeys(rel.getLeft, ignoreNulls)
      case _ =>
        val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.joinCondition)
        getJoinUniqueKeys(
          joinInfo, FlinkJoinRelType.toJoinRelType(rel.joinType), rel.getLeft, rel.getRight, mq,
          ignoreNulls)
    }
  }

  def getUniqueKeys(
      rel: BatchExecOverAggregate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] =
    getUniqueKeysOfOverWindow(rel, mq, ignoreNulls)

  def getUniqueKeys(
      rel: Window,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] =
    getUniqueKeysOfOverWindow(rel, mq, ignoreNulls)

  private def getUniqueKeysOfOverWindow(
      window: SingleRel,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] =
    mq.getUniqueKeys(window.getInput, ignoreNulls)

  // Catch-all rule when none of the others apply.
  def getUniqueKeys(
      rel: RelNode,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = null

  def getUniqueKeys(
      rel: RelSubset,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    if (!Bug.CALCITE_1048_FIXED) {
      //if the best node is null, so we can get the uniqueKeys based original node, due to
      //the original node is logically equivalent as the rel.
      mq.getUniqueKeys(Util.first(rel.getBest, rel.getOriginal), ignoreNulls)
    } else {
      throw new RuntimeException("CALCITE_1048 is fixed, so check this method again!")
    }
  }

  private def toImmutableSet(array: Array[Int]): util.Set[ImmutableBitSet] = {
    if (array.nonEmpty) {
      val keys = new JArrayList[Integer]()
      array.foreach(keys.add(_))
      ImmutableSet.of(ImmutableBitSet.of(keys))
    } else {
      null
    }
  }

  private def indexKeyToUniqueBitSet(
      index: Option[IndexKey]): util.Set[ImmutableBitSet] = index match {
    case Some(idx) if idx.isUnique =>
      toImmutableSet(idx.toArray)
    case _ => null
  }

  private def indexKeysToUniqueBitSet(
    indexes: util.Collection[IndexKey]): util.Set[ImmutableBitSet] = {
    if (null != indexes && indexes.nonEmpty) {
      val retSet = new util.HashSet[ImmutableBitSet]
      indexes.filter(_.isUnique).map(_.toArray).foreach { uk => retSet.addAll(toImmutableSet(uk)) }
      if (retSet.isEmpty) {
        null
      } else {
        retSet
      }
    } else {
      null
    }
  }

}

object FlinkRelMdUniqueKeys {

  private val INSTANCE = new FlinkRelMdUniqueKeys

  val SOURCE: RelMetadataProvider = ReflectiveRelMetadataProvider.reflectiveSource(
    BuiltInMethod.UNIQUE_KEYS.method, INSTANCE)

}