This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ab55f18cd37 Revert "Refactor RexNodeExtractor to use Scala collections"
ab55f18cd37 is described below
commit ab55f18cd37f320fcd1e875bc33348b7205ca6fd
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Mon Jun 29 10:54:08 2026 +0200
Revert "Refactor RexNodeExtractor to use Scala collections"
This reverts commit aebc8a42bd0ec82f177e49f2831d464da8c23dba.
---
.../planner/plan/utils/RexNodeExtractor.scala | 61 +++++++++-------------
1 file changed, 25 insertions(+), 36 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
index 8ebefbbdc95..4e2d1d66205 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
@@ -27,7 +27,6 @@ import org.apache.flink.table.expressions.ApiExpressionUtils._
import org.apache.flink.table.functions.{BuiltInFunctionDefinition,
FunctionIdentifier}
import org.apache.flink.table.functions.BuiltInFunctionDefinitions.{AND, CAST,
OR, TRY_CAST}
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
-import org.apache.flink.table.planner.codegen.CodeGenException
import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable
import org.apache.flink.table.planner.utils.Logging
import
org.apache.flink.table.planner.utils.TimestampStringUtils.toLocalDateTime
@@ -49,6 +48,7 @@ import java.time.{ZoneId, ZoneOffset}
import java.util
import java.util.{Collections, List => JList}
+import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
@@ -67,7 +67,7 @@ object RexNodeExtractor extends Logging {
def extractRefInputFields(exprs: JList[RexNode]): Array[Int] = {
val visitor = new InputRefVisitor
// extract referenced input fields from expressions
- exprs.asScala.foreach(_.accept(visitor))
+ exprs.foreach(_.accept(visitor))
visitor.getFields
}
@@ -86,7 +86,7 @@ object RexNodeExtractor extends Logging {
exprs: JList[RexNode],
usedFields: Array[Int]): Array[Array[JList[String]]] = {
val visitor = new RefFieldAccessorVisitor(usedFields)
- exprs.asScala.foreach(_.accept(visitor))
+ exprs.foreach(_.accept(visitor))
visitor.getProjectedFields
}
@@ -168,9 +168,8 @@ object RexNodeExtractor extends Logging {
partitionFieldNames: Array[String]): (RexNode, RexNode) = {
val (partitionPredicates, nonPartitionPredicates) =
extractPartitionPredicateList(expr, inputFieldNames, rexBuilder,
partitionFieldNames)
- val partitionPredicate = RexUtil.composeConjunction(rexBuilder,
partitionPredicates.asJava)
- val nonPartitionPredicate =
- RexUtil.composeConjunction(rexBuilder, nonPartitionPredicates.asJava)
+ val partitionPredicate = RexUtil.composeConjunction(rexBuilder,
partitionPredicates)
+ val nonPartitionPredicate = RexUtil.composeConjunction(rexBuilder,
nonPartitionPredicates)
(partitionPredicate, nonPartitionPredicate)
}
@@ -200,8 +199,7 @@ object RexNodeExtractor extends Logging {
val conjunctions = RelOptUtil.conjunctions(cnf)
val (partitionPredicates, nonPartitionPredicates) =
- conjunctions.asScala.toSeq.partition(
- isSupportedPartitionPredicate(_, partitionFieldNames, inputFieldNames))
+ conjunctions.partition(isSupportedPartitionPredicate(_,
partitionFieldNames, inputFieldNames))
(partitionPredicates, nonPartitionPredicates)
}
@@ -284,7 +282,7 @@ class InputRefVisitor extends RexVisitorImpl[Unit](true) {
fields += inputRef.getIndex
override def visitCall(call: RexCall): Unit =
- call.operands.asScala.foreach(operand => operand.accept(this))
+ call.operands.foreach(operand => operand.accept(this))
}
/** A RexVisitor to extract used nested input fields */
@@ -296,10 +294,10 @@ class RefFieldAccessorVisitor(usedFields: Array[Int])
extends RexVisitorImpl[Uni
private val order: Map[Int, Int] = usedFields.zipWithIndex.toMap
private def isPrefix(left: JList[String], right: JList[String]): Boolean = {
- if (right.size < left.size) {
+ if (right.length < left.length) {
false
} else {
- right.asScala.take(left.size).zip(left.asScala).foldLeft(true) {
+ right.take(left.length).zip(left).foldLeft(true) {
case (ans, (lName, rName)) => {
if (ans) {
lName.equals(rName)
@@ -321,21 +319,20 @@ class RefFieldAccessorVisitor(usedFields: Array[Int])
extends RexVisitorImpl[Uni
// get prefix field accesses
val prefixAccesses = sorted.foldLeft(Nil: List[JList[String]]) {
(prefixAccesses, nestedAccess) =>
- val nestedAccessJava: JList[String] = nestedAccess.asJava
prefixAccesses match {
// first access => add access
- case Nil => List[JList[String]](nestedAccessJava)
+ case Nil => List[JList[String]](nestedAccess)
// top-level access already found => return top-level access
case head :: Nil if head.get(0).equals("*") => prefixAccesses
// access is top-level access => return top-level access
- case _ :: _ if nestedAccessJava.get(0).equals("*") =>
List(util.Arrays.asList("*"))
+ case _ :: _ if nestedAccess.get(0).equals("*") =>
List(util.Arrays.asList("*"))
case _ =>
- if (isPrefix(prefixAccesses.head, nestedAccessJava)) {
+ if (isPrefix(prefixAccesses.head, nestedAccess)) {
// previous access is a prefix of this access => do not add
access
prefixAccesses
} else {
// previous access is not prefix of this access => add access
- nestedAccessJava :: prefixAccesses
+ nestedAccess :: prefixAccesses
}
}
}
@@ -373,7 +370,7 @@ class RefFieldAccessorVisitor(usedFields: Array[Int])
extends RexVisitorImpl[Uni
}
override def visitCall(call: RexCall): Unit =
- call.operands.asScala.foreach(operand => operand.accept(this))
+ call.operands.foreach(operand => operand.accept(this))
}
/**
@@ -508,7 +505,7 @@ class RexNodeToExpressionConverter(
override def visitCall(oriRexCall: RexCall): Option[ResolvedExpression] = {
val rexCall = FlinkRexUtil.expandSearch(rexBuilder,
oriRexCall).asInstanceOf[RexCall]
- val operands = rexCall.getOperands.asScala.toSeq.map(operand =>
operand.accept(this).orNull)
+ val operands = rexCall.getOperands.map(operand =>
operand.accept(this).orNull)
val outputType =
fromLogicalTypeToDataType(FlinkTypeFactory.toLogicalType(rexCall.getType))
@@ -518,21 +515,17 @@ class RexNodeToExpressionConverter(
} else {
rexCall.getOperator match {
case SqlStdOperatorTable.OR =>
- Option(
- operands.reduceLeft(
- (l, r) => CallExpression.permanent(OR, Seq(l, r).asJava,
outputType)))
+ Option(operands.reduceLeft((l, r) => CallExpression.permanent(OR,
Seq(l, r), outputType)))
case SqlStdOperatorTable.AND =>
Option(
- operands.reduceLeft(
- (l, r) => CallExpression.permanent(AND, Seq(l, r).asJava,
outputType)))
+ operands.reduceLeft((l, r) => CallExpression.permanent(AND, Seq(l,
r), outputType)))
case SqlStdOperatorTable.CAST =>
Option(
- CallExpression
- .permanent(CAST, Seq(operands.head,
typeLiteral(outputType)).asJava, outputType))
+ CallExpression.permanent(CAST, Seq(operands.head,
typeLiteral(outputType)), outputType))
case FlinkSqlOperatorTable.TRY_CAST =>
Option(
CallExpression
- .permanent(TRY_CAST, Seq(operands.head,
typeLiteral(outputType)).asJava, outputType))
+ .permanent(TRY_CAST, Seq(operands.head,
typeLiteral(outputType)), outputType))
case _: SqlFunction | _: SqlPostfixOperator =>
val names = new
util.ArrayList[String](rexCall.getOperator.getNameAsId.names)
names.set(names.size() - 1, replace(names.get(names.size() - 1)))
@@ -558,15 +551,15 @@ class RexNodeToExpressionConverter(
case Some(dataType) =>
val schema =
NestedProjectionUtil.build(Collections.singletonList(fieldAccess), dataType)
val fieldIndices = NestedProjectionUtil.convertToIndexArray(schema)
- var (topLevelColumnName, nestedColumn) = schema.columns.asScala.head
+ var (topLevelColumnName, nestedColumn) = schema.columns.head
val fieldNames = new ArrayBuffer[String]()
while (!nestedColumn.isLeaf) {
- fieldNames += topLevelColumnName
- topLevelColumnName = nestedColumn.children.asScala.head._1
- nestedColumn = nestedColumn.children.asScala.head._2
+ fieldNames.add(topLevelColumnName)
+ topLevelColumnName = nestedColumn.children.head._1
+ nestedColumn = nestedColumn.children.head._2
}
- fieldNames += topLevelColumnName
+ fieldNames.add(topLevelColumnName)
Some(
new NestedFieldReferenceExpression(
@@ -614,7 +607,7 @@ class RexNodeToExpressionConverter(
resolvedFunction.isTemporary,
identifier,
resolvedFunction.getDefinition,
- operands.asJava,
+ operands,
outputType)
)
} else {
@@ -628,8 +621,4 @@ class RexNodeToExpressionConverter(
str.replaceAll("\\s|_", "")
}
- override def visitNodeAndFieldIndex(
- nodeAndFieldIndex: RexNodeAndFieldIndex): Option[ResolvedExpression] = {
- throw new CodeGenException("RexNodeAndFieldIndex are not supported yet.")
- }
}
