You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2026/07/06 07:52:39 UTC
(flink) branch master updated (a6669809b9f -> 2d14af307fb)
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from a6669809b9f [FLINK-39957][table] Support `EXPLAIN` of `CREATE [OR ALTER] MATERIALIZED TABLE`
new 83506f9106a [FLINK-40062][table] Replace deprecated `JavaConversions` with `JavaConverters`
new 2d14af307fb [FLINK-40062][table] Allow `BridgingSqlAggFunction` being instantiated from `BuiltInFunctionDefinitions`, fix `$WELFORD_M2$1` function deserialization
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../rel/rules/AggregateReduceFunctionsRule.java | 6 +-
.../functions/bridging/BridgingSqlAggFunction.java | 21 +
.../functions/sql/FlinkSqlOperatorTable.java | 8 -
.../flink/table/planner/plan/utils/PythonUtil.java | 5 +-
.../planner/plan/utils/AggFunctionFactory.scala | 10 +-
...est.java => SortAggregateBatchRestoreTest.java} | 12 +-
.../exec/batch/SortAggregateTestPrograms.java | 62 +++
.../plan/sort-aggregate-variance.json} | 455 +++++++++------------
8 files changed, 282 insertions(+), 297 deletions(-)
copy flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/{LimitBatchRestoreTest.java => SortAggregateBatchRestoreTest.java} (76%)
create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/SortAggregateTestPrograms.java
copy flink-table/flink-table-planner/src/test/resources/restore-tests/{batch-exec-expand_1/expand/plan/expand.json => batch-exec-sort-aggregate_1/sort-aggregate-variance/plan/sort-aggregate-variance.json} (51%)
(flink) 02/02: [FLINK-40062][table] Allow `BridgingSqlAggFunction` being instantiated from `BuiltInFunctionDefinitions`, fix `$WELFORD_M2$1` function deserialization
Posted by sn...@apache.org.
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 2d14af307fb6efb2e4dfcdd19dd35ac69e9b44e1
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Fri Jul 3 12:59:52 2026 +0200
[FLINK-40062][table] Allow `BridgingSqlAggFunction` being instantiated from `BuiltInFunctionDefinitions`, fix `$WELFORD_M2$1` function deserialization
This closes #28628.
---
.../rel/rules/AggregateReduceFunctionsRule.java | 6 +-
.../functions/bridging/BridgingSqlAggFunction.java | 21 +
.../functions/sql/FlinkSqlOperatorTable.java | 8 -
.../flink/table/planner/plan/utils/PythonUtil.java | 5 +-
.../planner/plan/utils/AggFunctionFactory.scala | 6 +-
.../exec/batch/SortAggregateBatchRestoreTest.java | 37 ++
.../exec/batch/SortAggregateTestPrograms.java | 62 +++
.../plan/sort-aggregate-variance.json | 443 +++++++++++++++++++++
8 files changed, 573 insertions(+), 15 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/rules/AggregateReduceFunctionsRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/rules/AggregateReduceFunctionsRule.java
index 79819357250..c3fdf22c478 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/rules/AggregateReduceFunctionsRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/rules/AggregateReduceFunctionsRule.java
@@ -16,7 +16,8 @@
*/
package org.apache.calcite.rel.rules;
-import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlAggFunction;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
@@ -556,7 +557,8 @@ public class AggregateReduceFunctionsRule extends RelRule<AggregateReduceFunctio
final AggregateCall m2AggCall =
AggregateCall.create(
- FlinkSqlOperatorTable.INTERNAL_WELFORD_M2,
+ BridgingSqlAggFunction.of(
+ cluster, BuiltInFunctionDefinitions.INTERNAL_WELFORD_M2),
oldCall.isDistinct(),
oldCall.isApproximate(),
oldCall.ignoreNulls(),
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlAggFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlAggFunction.java
index cd6f57137bb..f0043361553 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlAggFunction.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlAggFunction.java
@@ -21,14 +21,18 @@ package org.apache.flink.table.planner.functions.bridging;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.catalog.ContextResolvedFunction;
import org.apache.flink.table.catalog.DataTypeFactory;
+import org.apache.flink.table.functions.BuiltInFunctionDefinition;
import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.FunctionIdentifier;
import org.apache.flink.table.functions.FunctionKind;
import org.apache.flink.table.functions.FunctionRequirement;
import org.apache.flink.table.planner.calcite.FlinkContext;
import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.TypeInference;
+import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.sql.SqlAggFunction;
import org.apache.calcite.sql.SqlKind;
@@ -126,6 +130,23 @@ public final class BridgingSqlAggFunction extends SqlAggFunction {
typeInference);
}
+ /** Creates an instance of an aggregating function during translation. */
+ public static BridgingSqlAggFunction of(
+ RelOptCluster cluster, ContextResolvedFunction resolvedFunction) {
+ final FlinkContext context = ShortcutUtils.unwrapContext(cluster);
+ final FlinkTypeFactory typeFactory = ShortcutUtils.unwrapTypeFactory(cluster);
+ return of(context, typeFactory, resolvedFunction);
+ }
+
+ /** Creates an instance of an aggregating built-in function during translation. */
+ public static BridgingSqlAggFunction of(
+ RelOptCluster cluster, BuiltInFunctionDefinition functionDefinition) {
+ return of(
+ cluster,
+ ContextResolvedFunction.permanent(
+ FunctionIdentifier.of(functionDefinition.getName()), functionDefinition));
+ }
+
public DataTypeFactory getDataTypeFactory() {
return dataTypeFactory;
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
index 7bcfd1e266e..32a2a20d679 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.planner.functions.sql;
import org.apache.flink.table.api.TableException;
-import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
import org.apache.flink.table.planner.functions.sql.internal.SqlAuxiliaryGroupAggFunction;
import org.apache.flink.table.planner.functions.sql.ml.SqlMLEvaluateTableFunction;
@@ -1116,13 +1115,6 @@ public class FlinkSqlOperatorTable extends ReflectiveSqlOperatorTable {
public static final SqlAggFunction VARIANCE = SqlStdOperatorTable.VARIANCE;
public static final SqlAggFunction VAR_POP = SqlStdOperatorTable.VAR_POP;
public static final SqlAggFunction VAR_SAMP = SqlStdOperatorTable.VAR_SAMP;
- public static final SqlAggFunction INTERNAL_WELFORD_M2 =
- SqlBasicAggFunction.create(
- BuiltInFunctionDefinitions.INTERNAL_WELFORD_M2.getName(),
- SqlKind.OTHER_FUNCTION,
- ReturnTypes.DOUBLE.andThen(SqlTypeTransforms.FORCE_NULLABLE),
- OperandTypes.NUMERIC)
- .withFunctionType(SqlFunctionCategory.SYSTEM);
public static final SqlAggFunction SINGLE_VALUE = SqlStdOperatorTable.SINGLE_VALUE;
public static final SqlAggFunction APPROX_COUNT_DISTINCT =
SqlStdOperatorTable.APPROX_COUNT_DISTINCT;
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/PythonUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/PythonUtil.java
index 65efb05f430..8691b30cd0d 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/PythonUtil.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/PythonUtil.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.utils;
+import org.apache.flink.table.functions.BuiltInFunctionDefinition;
import org.apache.flink.table.functions.DeclarativeAggregateFunction;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.functions.python.PythonFunction;
@@ -141,7 +142,9 @@ public class PythonUtil {
return aggSqlFunction.aggregateFunction() instanceof BuiltInAggregateFunction;
} else if (aggregation instanceof BridgingSqlAggFunction) {
BridgingSqlAggFunction bridgingSqlAggFunction = (BridgingSqlAggFunction) aggregation;
- return bridgingSqlAggFunction.getDefinition() instanceof DeclarativeAggregateFunction;
+ FunctionDefinition definition = bridgingSqlAggFunction.getDefinition();
+ return definition instanceof DeclarativeAggregateFunction
+ || definition instanceof BuiltInFunctionDefinition;
} else {
return true;
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala
index a043560cd68..9b4fc9d811e 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala
@@ -142,10 +142,6 @@ class AggFunctionFactory(
case _: SqlListAggFunction if call.getArgList.size() == 2 =>
createListAggWsFunction(argTypes, index)
- case a: SqlBasicAggFunction
- if a.getName == BuiltInFunctionDefinitions.INTERNAL_WELFORD_M2.getName =>
- createWelfordM2AggFunction(argTypes)
-
// TODO supports SqlCardinalityCountAggFunction
case a: SqlAggFunction if a.getKind == SqlKind.COLLECT =>
@@ -172,6 +168,8 @@ class AggFunctionFactory(
case bridge: BridgingSqlAggFunction =>
bridge.getDefinition match {
// built-in imperativeFunction
+ case BuiltInFunctionDefinitions.INTERNAL_WELFORD_M2 =>
+ createWelfordM2AggFunction(argTypes)
case BuiltInFunctionDefinitions.PERCENTILE =>
createPercentileAggFunction(argTypes)
case BuiltInFunctionDefinitions.BITMAP_BUILD_AGG =>
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/SortAggregateBatchRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/SortAggregateBatchRestoreTest.java
new file mode 100644
index 00000000000..d329a2d04ec
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/SortAggregateBatchRestoreTest.java
@@ -0,0 +1,37 @@
+/*
+ * 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.nodes.exec.batch;
+
+import org.apache.flink.table.planner.plan.nodes.exec.testutils.BatchRestoreTestBase;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.List;
+
+/** Batch Compiled Plan tests for {@link BatchExecSortAggregate}. */
+class SortAggregateBatchRestoreTest extends BatchRestoreTestBase {
+
+ public SortAggregateBatchRestoreTest() {
+ super(BatchExecSortAggregate.class);
+ }
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return List.of(SortAggregateTestPrograms.GROUP_BY_VARIANCE);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/SortAggregateTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/SortAggregateTestPrograms.java
new file mode 100644
index 00000000000..89aa5a333a6
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/SortAggregateTestPrograms.java
@@ -0,0 +1,62 @@
+/*
+ * 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.nodes.exec.batch;
+
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+
+/** {@link TableTestProgram} definitions for testing {@link BatchExecSortAggregate}. */
+class SortAggregateTestPrograms {
+
+ static final TableTestProgram GROUP_BY_VARIANCE =
+ TableTestProgram.of(
+ "sort-aggregate-variance",
+ "validates variance related aggregations (VAR_POP, VAR_SAMP) that are "
+ + "reduced to the internal $WELFORD_M2$1 aggregate function")
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema("a INT", "b BIGINT")
+ .producedBeforeRestore(
+ Row.of(2, 1L),
+ Row.of(4, 1L),
+ Row.of(10, 2L),
+ Row.of(14, 2L))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "b BIGINT",
+ "cnt BIGINT",
+ "var_pop_a DOUBLE",
+ "var_samp_a DOUBLE",
+ "PRIMARY KEY (b) NOT ENFORCED")
+ .consumedBeforeRestore(
+ "+I[1, 2, 1.0, 2.0]", "+I[2, 2, 4.0, 8.0]")
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT "
+ + "b, "
+ + "COUNT(*) AS cnt, "
+ + "VAR_POP(a) AS var_pop_a, "
+ + "VAR_SAMP(a) AS var_samp_a "
+ + "FROM source_t GROUP BY b")
+ .build();
+}
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-sort-aggregate_1/sort-aggregate-variance/plan/sort-aggregate-variance.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-sort-aggregate_1/sort-aggregate-variance/plan/sort-aggregate-variance.json
new file mode 100644
index 00000000000..eaef8b633a7
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-sort-aggregate_1/sort-aggregate-variance/plan/sort-aggregate-variance.json
@@ -0,0 +1,443 @@
+{
+ "flinkVersion" : "2.4",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`a` INT, `b` BIGINT>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, source_t]], fields=[a, b])",
+ "dynamicFilteringDataListenerID" : "f7b00c3e-afdd-4dfc-a6db-479041ae639e"
+ }, {
+ "id" : 2,
+ "type" : "batch-exec-sort_1",
+ "configuration" : {
+ "table.exec.resource.sort.memory" : "128 mb",
+ "table.exec.sort.async-merge-enabled" : "true",
+ "table.exec.sort.max-num-file-handles" : "128",
+ "table.exec.spill-compression.block-size" : "64 kb",
+ "table.exec.spill-compression.enabled" : "true"
+ },
+ "sortSpec" : {
+ "fields" : [ {
+ "index" : 1,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT>",
+ "description" : "Sort(orderBy=[b ASC])"
+ }, {
+ "id" : 9,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "KEEP_INPUT_AS_IS",
+ "inputDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "isStrict" : true
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT>",
+ "description" : "Exchange(distribution=[forward])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 3,
+ "type" : "batch-exec-sort-aggregate_1",
+ "grouping" : [ 1 ],
+ "auxGrouping" : [ ],
+ "aggCalls" : [ {
+ "name" : "cnt",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : null,
+ "internalName" : "$WELFORD_M2$1",
+ "argList" : [ 0 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "DOUBLE"
+ }, {
+ "name" : null,
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 0 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "aggInputRowType" : "ROW<`a` INT, `b` BIGINT>",
+ "isMerge" : false,
+ "isFinal" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `count1$0` BIGINT, `$f2` STRUCTURED<'org.apache.flink.table.runtime.functions.aggregate.WelfordM2AggFunction$WelfordM2Accumulator', `n` BIGINT NOT NULL, `mean` DOUBLE NOT NULL, `m2` DOUBLE NOT NULL>, `count$1` BIGINT>",
+ "description" : "LocalSortAggregate(groupBy=[b], select=[b, Partial_COUNT(*) AS count1$0, Partial_$WELFORD_M2$1(a) AS $f2, Partial_COUNT(a) AS count$1])"
+ }, {
+ "id" : 4,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `count1$0` BIGINT, `$f2` STRUCTURED<'org.apache.flink.table.runtime.functions.aggregate.WelfordM2AggFunction$WelfordM2Accumulator', `n` BIGINT NOT NULL, `mean` DOUBLE NOT NULL, `m2` DOUBLE NOT NULL>, `count$1` BIGINT>",
+ "description" : "Exchange(distribution=[hash[b]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 5,
+ "type" : "batch-exec-sort_1",
+ "configuration" : {
+ "table.exec.resource.sort.memory" : "128 mb",
+ "table.exec.sort.async-merge-enabled" : "true",
+ "table.exec.sort.max-num-file-handles" : "128",
+ "table.exec.spill-compression.block-size" : "64 kb",
+ "table.exec.spill-compression.enabled" : "true"
+ },
+ "sortSpec" : {
+ "fields" : [ {
+ "index" : 0,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `count1$0` BIGINT, `$f2` STRUCTURED<'org.apache.flink.table.runtime.functions.aggregate.WelfordM2AggFunction$WelfordM2Accumulator', `n` BIGINT NOT NULL, `mean` DOUBLE NOT NULL, `m2` DOUBLE NOT NULL>, `count$1` BIGINT>",
+ "description" : "Sort(orderBy=[b ASC])"
+ }, {
+ "id" : 10,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "KEEP_INPUT_AS_IS",
+ "inputDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "isStrict" : true
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `count1$0` BIGINT, `$f2` STRUCTURED<'org.apache.flink.table.runtime.functions.aggregate.WelfordM2AggFunction$WelfordM2Accumulator', `n` BIGINT NOT NULL, `mean` DOUBLE NOT NULL, `m2` DOUBLE NOT NULL>, `count$1` BIGINT>",
+ "description" : "Exchange(distribution=[forward])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 6,
+ "type" : "batch-exec-sort-aggregate_1",
+ "grouping" : [ 0 ],
+ "auxGrouping" : [ ],
+ "aggCalls" : [ {
+ "name" : "cnt",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : null,
+ "internalName" : "$WELFORD_M2$1",
+ "argList" : [ 0 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "DOUBLE"
+ }, {
+ "name" : null,
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 0 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "aggInputRowType" : "ROW<`a` INT, `b` BIGINT>",
+ "isMerge" : true,
+ "isFinal" : true,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `cnt` BIGINT NOT NULL, `$f2` DOUBLE, `$f3` BIGINT NOT NULL>",
+ "description" : "SortAggregate(isMerge=[true], groupBy=[b], select=[b, Final_COUNT(count1$0) AS cnt, Final_$WELFORD_M2$1($f2) AS $f2, Final_COUNT(count$1) AS $f3])"
+ }, {
+ "id" : 7,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$/$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "DOUBLE"
+ } ],
+ "type" : "INT"
+ } ],
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$/$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CASE$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : null,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$-$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ } ],
+ "type" : "DOUBLE"
+ } ],
+ "type" : "INT"
+ } ],
+ "type" : "DOUBLE"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `cnt` BIGINT, `var_pop_a` DOUBLE, `var_samp_a` DOUBLE>",
+ "description" : "Calc(select=[b, CAST(cnt AS BIGINT) AS cnt, CAST(CAST(($f2 / $f3) AS INTEGER) AS DOUBLE) AS var_pop_a, CAST(CAST(($f2 / CASE(($f3 = 1), null:BIGINT, ($f3 - 1))) AS INTEGER) AS DOUBLE) AS var_samp_a])"
+ }, {
+ "id" : 8,
+ "type" : "batch-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.type-length-enforcer" : "IGNORE"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "b",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "cnt",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "var_pop_a",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "var_samp_a",
+ "dataType" : "DOUBLE"
+ } ],
+ "primaryKey" : {
+ "name" : "PK_b",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "b" ]
+ }
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `cnt` BIGINT, `var_pop_a` DOUBLE, `var_samp_a` DOUBLE>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[b, cnt, var_pop_a, var_samp_a])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 9,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 9,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 5,
+ "target" : 10,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 10,
+ "target" : 6,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 6,
+ "target" : 7,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 7,
+ "target" : 8,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
(flink) 01/02: [FLINK-40062][table] Replace deprecated `JavaConversions` with `JavaConverters`
Posted by sn...@apache.org.
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 83506f9106a7ee9fab53004b2279f71cf9fa6fa2
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Sat Jul 4 12:30:14 2026 +0200
[FLINK-40062][table] Replace deprecated `JavaConversions` with `JavaConverters`
---
.../apache/flink/table/planner/plan/utils/AggFunctionFactory.scala | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala
index 37ba914bacd..a043560cd68 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala
@@ -37,7 +37,7 @@ import org.apache.calcite.sql.fun._
import java.util
-import scala.collection.JavaConversions._
+import scala.collection.JavaConverters._
/**
* Factory for creating runtime implementation for internal aggregate functions that are declared as
@@ -62,7 +62,7 @@ class AggFunctionFactory(
/** The entry point to create an aggregate function from the given [[AggregateCall]]. */
def createAggFunction(call: AggregateCall, index: Int): UserDefinedFunction = {
- val argTypes: Array[LogicalType] = call.getArgList
+ val argTypes: Array[LogicalType] = call.getArgList.asScala
.map(inputRowType.getChildren.get(_))
.toArray