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
commit fb9c511f5901fe653dee080c58a3e0748b4a569d Author: Sergey Nuyanzin <[email protected]> AuthorDate: Mon Jun 29 10:28:12 2026 +0200 [FLINK-39859][table] Replace `Javaconverions` with `JavaConverters` in `RemoteCalcSplitRule` and `RemoteCalcSpliRule` --- .../plan/rules/logical/RemoteCalcSplitRule.scala | 43 ++++++++--------- .../planner/plan/utils/RexNodeExtractor.scala | 56 ++++++++++++---------- 2 files changed, 52 insertions(+), 47 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcSplitRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcSplitRule.scala index 26baa44f43d..f4537225dbc 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcSplitRule.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcSplitRule.scala @@ -28,7 +28,6 @@ import org.apache.calcite.sql.validate.SqlValidatorUtil import java.util.function.Function -import scala.collection.JavaConversions._ import scala.collection.JavaConverters._ import scala.collection.mutable @@ -94,8 +93,8 @@ abstract class RemoteCalcSplitRuleBase[T]( val bottomCalcProjects = accessedFields.map(RexInputRef.of(_, input.getRowType)) ++ extractedRexNodes val bottomCalcFieldNames = SqlValidatorUtil.uniquify( - accessedFields.map(i => input.getRowType.getFieldNames.get(i)).toSeq ++ - extractedRexNodes.indices.map("f" + _), + (accessedFields.map(i => input.getRowType.getFieldNames.get(i)).toSeq ++ + extractedRexNodes.indices.map("f" + _)).asJava, rexBuilder.getTypeFactory.getTypeSystem.isSchemaCaseSensitive ) @@ -105,7 +104,7 @@ abstract class RemoteCalcSplitRuleBase[T]( input, RexProgram.create( input.getRowType, - bottomCalcProjects.toList, + bottomCalcProjects.toList.asJava, splitComponents.bottomCalcCondition.orNull, bottomCalcFieldNames, rexBuilder)) @@ -120,7 +119,7 @@ abstract class RemoteCalcSplitRuleBase[T]( bottomCalc, RexProgram.create( bottomCalc.getRowType, - topCalcProjects.map(_.accept(inputRewriter)), + topCalcProjects.map(_.accept(inputRewriter)).asJava, splitComponents.topCalcCondition.map(_.accept(inputRewriter)).orNull, calc.getRowType, rexBuilder @@ -204,7 +203,7 @@ class RemoteCalcSplitConditionRule(callFinder: RemoteCallFinder) new SplitComponents( None, Option(program.getCondition).map(program.expandLocalRef(_).accept(splitter)), - program.getProjectList.map(program.expandLocalRef)) + program.getProjectList.asScala.toSeq.map(program.expandLocalRef)) } } @@ -217,7 +216,7 @@ abstract class RemoteCalcSplitProjectionRuleBase[T]( new SplitComponents( Option(program.getCondition).map(program.expandLocalRef), None, - program.getProjectList.map(program.expandLocalRef(_).accept(splitter))) + program.getProjectList.asScala.toSeq.map(program.expandLocalRef(_).accept(splitter))) } } @@ -240,7 +239,7 @@ abstract class RemoteCalcSplitRexFieldRuleBase(description: String, callFinder: protected def containsFieldAccessAfterRemoteCall(node: RexNode): Boolean = { node match { - case call: RexCall => call.getOperands.exists(containsFieldAccessAfterRemoteCall) + case call: RexCall => call.getOperands.asScala.exists(containsFieldAccessAfterRemoteCall) case x: RexFieldAccess => callFinder.containsRemoteCall(x.getReferenceExpr) case _ => false } @@ -257,7 +256,7 @@ class RemoteCalcSplitProjectionRexFieldRule(callFinder: RemoteCallFinder) override def matches(call: RelOptRuleCall): Boolean = { val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc] - val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef) + val projects = calc.getProgram.getProjectList.asScala.toSeq.map(calc.getProgram.expandLocalRef) projects.exists(containsFieldAccessAfterRemoteCall) } @@ -265,7 +264,7 @@ class RemoteCalcSplitProjectionRexFieldRule(callFinder: RemoteCallFinder) new SplitComponents( Option(program.getCondition).map(program.expandLocalRef), None, - program.getProjectList.map(_.accept(splitter))) + program.getProjectList.asScala.toSeq.map(_.accept(splitter))) } } @@ -288,7 +287,7 @@ class RemoteCalcSplitConditionRexFieldRule(callFinder: RemoteCallFinder) new SplitComponents( None, Option(program.getCondition).map(_.accept(splitter)), - program.getProjectList.map(_.accept(splitter))) + program.getProjectList.asScala.toSeq.map(_.accept(splitter))) } } @@ -302,7 +301,7 @@ class RemoteCalcSplitProjectionRule(callFinder: RemoteCallFinder) override def matches(call: RelOptRuleCall): Boolean = { val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc] - val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef) + val projects = calc.getProgram.getProjectList.asScala.toSeq.map(calc.getProgram.expandLocalRef) // matches if it contains both Remote functions and Java functions in the projection projects.exists(callFinder.containsRemoteCall) && projects.exists( @@ -313,7 +312,7 @@ class RemoteCalcSplitProjectionRule(callFinder: RemoteCallFinder) program: RexProgram, node: RexNode, matchState: Option[Nothing]): Boolean = { - program.getProjectList + program.getProjectList.asScala.toSeq .map(program.expandLocalRef) .exists(callFinder.isNonRemoteCall) == callFinder.isRemoteCall(node) } @@ -328,7 +327,7 @@ class RemoteCalcExpandProjectRule(callFinder: RemoteCallFinder) override def matches(call: RelOptRuleCall): Boolean = { val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc] - val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef) + val projects = calc.getProgram.getProjectList.asScala.toSeq.map(calc.getProgram.expandLocalRef) projects.exists(callFinder.containsRemoteCall) && projects.exists(containsFieldAccessInputs) } @@ -343,12 +342,12 @@ class RemoteCalcExpandProjectRule(callFinder: RemoteCallFinder) new SplitComponents( Option(program.getCondition).map(program.expandLocalRef), None, - program.getProjectList.map(program.expandLocalRef(_).accept(splitter))) + program.getProjectList.asScala.toSeq.map(program.expandLocalRef(_).accept(splitter))) } private def containsFieldAccessInputs(node: RexNode): Boolean = { node match { - case call: RexCall => call.getOperands.exists(containsFieldAccessInputs) + case call: RexCall => call.getOperands.asScala.exists(containsFieldAccessInputs) case _: RexFieldAccess => true case _ => false } @@ -364,7 +363,7 @@ class RemoteCalcPushConditionRule(callFinder: RemoteCallFinder) override def matches(call: RelOptRuleCall): Boolean = { val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc] - val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef) + val projects = calc.getProgram.getProjectList.asScala.toSeq.map(calc.getProgram.expandLocalRef) // matches if all the following conditions hold true: // 1) the condition is not null @@ -382,7 +381,7 @@ class RemoteCalcPushConditionRule(callFinder: RemoteCallFinder) new SplitComponents( Option(program.getCondition).map(program.expandLocalRef), None, - program.getProjectList.map(program.expandLocalRef)) + program.getProjectList.asScala.toSeq.map(program.expandLocalRef)) } } @@ -397,7 +396,7 @@ class RemoteCalcRewriteProjectionRule(callFinder: RemoteCallFinder) override def matches(call: RelOptRuleCall): Boolean = { val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc] - val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef) + val projects = calc.getProgram.getProjectList.asScala.toSeq.map(calc.getProgram.expandLocalRef) // matches if all the following conditions hold true: // 1) it contains Remote functions in the projection @@ -419,7 +418,7 @@ class RemoteCalcRewriteProjectionRule(callFinder: RemoteCallFinder) new SplitComponents( None, None, - program.getProjectList.map(program.expandLocalRef(_).accept(splitter))) + program.getProjectList.asScala.toSeq.map(program.expandLocalRef(_).accept(splitter))) } } @@ -440,7 +439,7 @@ class ScalarFunctionSplitter( if (needConvert(call)) { getExtractedRexNode(call) } else { - call.clone(call.getType, call.getOperands.asScala.map(_.accept(this))) + call.clone(call.getType, call.getOperands.asScala.map(_.accept(this)).asJava) } } @@ -536,7 +535,7 @@ private class ExtractedFunctionInputRewriter( } override def visitCall(call: RexCall): RexNode = { - call.clone(call.getType, call.getOperands.asScala.map(_.accept(this))) + call.clone(call.getType, call.getOperands.asScala.map(_.accept(this)).asJava) } override def visitFieldAccess(fieldAccess: RexFieldAccess): RexNode = { 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 4e2d1d66205..59c1ae9e2e3 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 @@ -48,7 +48,6 @@ 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 +66,7 @@ object RexNodeExtractor extends Logging { def extractRefInputFields(exprs: JList[RexNode]): Array[Int] = { val visitor = new InputRefVisitor // extract referenced input fields from expressions - exprs.foreach(_.accept(visitor)) + exprs.asScala.foreach(_.accept(visitor)) visitor.getFields } @@ -86,7 +85,7 @@ object RexNodeExtractor extends Logging { exprs: JList[RexNode], usedFields: Array[Int]): Array[Array[JList[String]]] = { val visitor = new RefFieldAccessorVisitor(usedFields) - exprs.foreach(_.accept(visitor)) + exprs.asScala.foreach(_.accept(visitor)) visitor.getProjectedFields } @@ -168,8 +167,9 @@ object RexNodeExtractor extends Logging { partitionFieldNames: Array[String]): (RexNode, RexNode) = { val (partitionPredicates, nonPartitionPredicates) = extractPartitionPredicateList(expr, inputFieldNames, rexBuilder, partitionFieldNames) - val partitionPredicate = RexUtil.composeConjunction(rexBuilder, partitionPredicates) - val nonPartitionPredicate = RexUtil.composeConjunction(rexBuilder, nonPartitionPredicates) + val partitionPredicate = RexUtil.composeConjunction(rexBuilder, partitionPredicates.asJava) + val nonPartitionPredicate = + RexUtil.composeConjunction(rexBuilder, nonPartitionPredicates.asJava) (partitionPredicate, nonPartitionPredicate) } @@ -199,7 +199,8 @@ object RexNodeExtractor extends Logging { val conjunctions = RelOptUtil.conjunctions(cnf) val (partitionPredicates, nonPartitionPredicates) = - conjunctions.partition(isSupportedPartitionPredicate(_, partitionFieldNames, inputFieldNames)) + conjunctions.asScala.toSeq.partition( + isSupportedPartitionPredicate(_, partitionFieldNames, inputFieldNames)) (partitionPredicates, nonPartitionPredicates) } @@ -282,7 +283,7 @@ class InputRefVisitor extends RexVisitorImpl[Unit](true) { fields += inputRef.getIndex override def visitCall(call: RexCall): Unit = - call.operands.foreach(operand => operand.accept(this)) + call.operands.asScala.foreach(operand => operand.accept(this)) } /** A RexVisitor to extract used nested input fields */ @@ -294,10 +295,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.length < left.length) { + if (right.size < left.size) { false } else { - right.take(left.length).zip(left).foldLeft(true) { + right.asScala.take(left.size).zip(left.asScala).foldLeft(true) { case (ans, (lName, rName)) => { if (ans) { lName.equals(rName) @@ -319,20 +320,21 @@ 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]](nestedAccess) + case Nil => List[JList[String]](nestedAccessJava) // 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 nestedAccess.get(0).equals("*") => List(util.Arrays.asList("*")) + case _ :: _ if nestedAccessJava.get(0).equals("*") => List(util.Arrays.asList("*")) case _ => - if (isPrefix(prefixAccesses.head, nestedAccess)) { + if (isPrefix(prefixAccesses.head, nestedAccessJava)) { // previous access is a prefix of this access => do not add access prefixAccesses } else { // previous access is not prefix of this access => add access - nestedAccess :: prefixAccesses + nestedAccessJava :: prefixAccesses } } } @@ -370,7 +372,7 @@ class RefFieldAccessorVisitor(usedFields: Array[Int]) extends RexVisitorImpl[Uni } override def visitCall(call: RexCall): Unit = - call.operands.foreach(operand => operand.accept(this)) + call.operands.asScala.foreach(operand => operand.accept(this)) } /** @@ -505,7 +507,7 @@ class RexNodeToExpressionConverter( override def visitCall(oriRexCall: RexCall): Option[ResolvedExpression] = { val rexCall = FlinkRexUtil.expandSearch(rexBuilder, oriRexCall).asInstanceOf[RexCall] - val operands = rexCall.getOperands.map(operand => operand.accept(this).orNull) + val operands = rexCall.getOperands.asScala.toSeq.map(operand => operand.accept(this).orNull) val outputType = fromLogicalTypeToDataType(FlinkTypeFactory.toLogicalType(rexCall.getType)) @@ -515,17 +517,21 @@ class RexNodeToExpressionConverter( } else { rexCall.getOperator match { case SqlStdOperatorTable.OR => - Option(operands.reduceLeft((l, r) => CallExpression.permanent(OR, Seq(l, r), outputType))) + Option( + operands.reduceLeft( + (l, r) => CallExpression.permanent(OR, Seq(l, r).asJava, outputType))) case SqlStdOperatorTable.AND => Option( - operands.reduceLeft((l, r) => CallExpression.permanent(AND, Seq(l, r), outputType))) + operands.reduceLeft( + (l, r) => CallExpression.permanent(AND, Seq(l, r).asJava, outputType))) case SqlStdOperatorTable.CAST => Option( - CallExpression.permanent(CAST, Seq(operands.head, typeLiteral(outputType)), outputType)) + CallExpression + .permanent(CAST, Seq(operands.head, typeLiteral(outputType)).asJava, outputType)) case FlinkSqlOperatorTable.TRY_CAST => Option( CallExpression - .permanent(TRY_CAST, Seq(operands.head, typeLiteral(outputType)), outputType)) + .permanent(TRY_CAST, Seq(operands.head, typeLiteral(outputType)).asJava, 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))) @@ -551,15 +557,15 @@ class RexNodeToExpressionConverter( case Some(dataType) => val schema = NestedProjectionUtil.build(Collections.singletonList(fieldAccess), dataType) val fieldIndices = NestedProjectionUtil.convertToIndexArray(schema) - var (topLevelColumnName, nestedColumn) = schema.columns.head + var (topLevelColumnName, nestedColumn) = schema.columns.asScala.head val fieldNames = new ArrayBuffer[String]() while (!nestedColumn.isLeaf) { - fieldNames.add(topLevelColumnName) - topLevelColumnName = nestedColumn.children.head._1 - nestedColumn = nestedColumn.children.head._2 + fieldNames += topLevelColumnName + topLevelColumnName = nestedColumn.children.asScala.head._1 + nestedColumn = nestedColumn.children.asScala.head._2 } - fieldNames.add(topLevelColumnName) + fieldNames += topLevelColumnName Some( new NestedFieldReferenceExpression( @@ -607,7 +613,7 @@ class RexNodeToExpressionConverter( resolvedFunction.isTemporary, identifier, resolvedFunction.getDefinition, - operands, + operands.asJava, outputType) ) } else {
