This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new f73dec412d7 [FLINK-40062][table] Allow `BridgingSqlAggFunction` being
instantiated from `BuiltInFunctionDefinitions`, fix `$WELFORD_M2$1` function
deserialization (#28647)
f73dec412d7 is described below
commit f73dec412d741d8c525ba747fae050e9bf81b1a7
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Mon Jul 6 16:53:50 2026 +0200
[FLINK-40062][table] Allow `BridgingSqlAggFunction` being instantiated from
`BuiltInFunctionDefinitions`, fix `$WELFORD_M2$1` function deserialization
(#28647)
---
.../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 817284846d1..d0a5097aa85 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;
@@ -1153,13 +1152,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 37ba914bacd..363379c0af3 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..d8fe2e65d7e
--- /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.3",
+ "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"
+ } ]
+}
