[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r378646800
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/validate/ProcedureNamespace.java
##
@@ -0,0 +1,80 @@
+/*
+ * 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.calcite.sql.validate;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.functions.utils.TableSqlFunction;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlUtil;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+/**
+ * Namespace whose contents are defined by the result of a call to a
user-defined procedure.
+ *
+ * Note: Compared to Calcite, this class implements custom logic for
dealing with collection tables
+ * like {@code TABLE(function(...))} procedures. Compared to the SQL standard,
Flink's table functions
+ * can return arbitrary types that are wrapped into a ROW type if necessary.
We don't interpret ARRAY
+ * or MULTISET types as it would be standard.
+ */
+@Internal
+public final class ProcedureNamespace extends AbstractNamespace {
+
+ private final SqlValidatorScope scope;
+
+ private final SqlCall call;
+
+ ProcedureNamespace(
+ SqlValidatorImpl validator,
+ SqlValidatorScope scope,
+ SqlCall call,
+ SqlNode enclosingNode) {
+ super(validator, enclosingNode);
+ this.scope = scope;
+ this.call = call;
+ }
+
+ public RelDataType validateImpl(RelDataType targetRowType) {
+ validator.inferUnknownTypes(validator.unknownType, scope, call);
+ final RelDataType type = validator.deriveTypeImpl(scope, call);
+ final SqlOperator operator = call.getOperator();
+ final SqlCallBinding callBinding =
+ new SqlCallBinding(validator, scope, call);
+ // legacy table functions
+ if (operator instanceof TableSqlFunction) {
Review comment:
It seems that we can not use `TableSqlFunction`, because `TableSqlFunction`s
are different between legacy planner and blink planner.
This lead to we have different `ProcedureNamespace` codes between legacy
planner and blink planner.
This will lead to bugs.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r378646800
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/validate/ProcedureNamespace.java
##
@@ -0,0 +1,80 @@
+/*
+ * 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.calcite.sql.validate;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.functions.utils.TableSqlFunction;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlUtil;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+/**
+ * Namespace whose contents are defined by the result of a call to a
user-defined procedure.
+ *
+ * Note: Compared to Calcite, this class implements custom logic for
dealing with collection tables
+ * like {@code TABLE(function(...))} procedures. Compared to the SQL standard,
Flink's table functions
+ * can return arbitrary types that are wrapped into a ROW type if necessary.
We don't interpret ARRAY
+ * or MULTISET types as it would be standard.
+ */
+@Internal
+public final class ProcedureNamespace extends AbstractNamespace {
+
+ private final SqlValidatorScope scope;
+
+ private final SqlCall call;
+
+ ProcedureNamespace(
+ SqlValidatorImpl validator,
+ SqlValidatorScope scope,
+ SqlCall call,
+ SqlNode enclosingNode) {
+ super(validator, enclosingNode);
+ this.scope = scope;
+ this.call = call;
+ }
+
+ public RelDataType validateImpl(RelDataType targetRowType) {
+ validator.inferUnknownTypes(validator.unknownType, scope, call);
+ final RelDataType type = validator.deriveTypeImpl(scope, call);
+ final SqlOperator operator = call.getOperator();
+ final SqlCallBinding callBinding =
+ new SqlCallBinding(validator, scope, call);
+ // legacy table functions
+ if (operator instanceof TableSqlFunction) {
Review comment:
It seems that we can use `TableSqlFunction`, because `TableSqlFunction`s are
different between legacy planner and blink planner.
This lead to we have different `ProcedureNamespace` codes in between legacy
planner and blink planner.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r378646800
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/validate/ProcedureNamespace.java
##
@@ -0,0 +1,80 @@
+/*
+ * 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.calcite.sql.validate;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.functions.utils.TableSqlFunction;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlUtil;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+/**
+ * Namespace whose contents are defined by the result of a call to a
user-defined procedure.
+ *
+ * Note: Compared to Calcite, this class implements custom logic for
dealing with collection tables
+ * like {@code TABLE(function(...))} procedures. Compared to the SQL standard,
Flink's table functions
+ * can return arbitrary types that are wrapped into a ROW type if necessary.
We don't interpret ARRAY
+ * or MULTISET types as it would be standard.
+ */
+@Internal
+public final class ProcedureNamespace extends AbstractNamespace {
+
+ private final SqlValidatorScope scope;
+
+ private final SqlCall call;
+
+ ProcedureNamespace(
+ SqlValidatorImpl validator,
+ SqlValidatorScope scope,
+ SqlCall call,
+ SqlNode enclosingNode) {
+ super(validator, enclosingNode);
+ this.scope = scope;
+ this.call = call;
+ }
+
+ public RelDataType validateImpl(RelDataType targetRowType) {
+ validator.inferUnknownTypes(validator.unknownType, scope, call);
+ final RelDataType type = validator.deriveTypeImpl(scope, call);
+ final SqlOperator operator = call.getOperator();
+ final SqlCallBinding callBinding =
+ new SqlCallBinding(validator, scope, call);
+ // legacy table functions
+ if (operator instanceof TableSqlFunction) {
Review comment:
It seems that we can use `TableSqlFunction`, because `TableSqlFunction`s are
different between legacy planner and blink planner.
This lead to we have different `ProcedureNamespace` codes between legacy
planner and blink planner.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r378646800
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/validate/ProcedureNamespace.java
##
@@ -0,0 +1,80 @@
+/*
+ * 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.calcite.sql.validate;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.functions.utils.TableSqlFunction;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlUtil;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+/**
+ * Namespace whose contents are defined by the result of a call to a
user-defined procedure.
+ *
+ * Note: Compared to Calcite, this class implements custom logic for
dealing with collection tables
+ * like {@code TABLE(function(...))} procedures. Compared to the SQL standard,
Flink's table functions
+ * can return arbitrary types that are wrapped into a ROW type if necessary.
We don't interpret ARRAY
+ * or MULTISET types as it would be standard.
+ */
+@Internal
+public final class ProcedureNamespace extends AbstractNamespace {
+
+ private final SqlValidatorScope scope;
+
+ private final SqlCall call;
+
+ ProcedureNamespace(
+ SqlValidatorImpl validator,
+ SqlValidatorScope scope,
+ SqlCall call,
+ SqlNode enclosingNode) {
+ super(validator, enclosingNode);
+ this.scope = scope;
+ this.call = call;
+ }
+
+ public RelDataType validateImpl(RelDataType targetRowType) {
+ validator.inferUnknownTypes(validator.unknownType, scope, call);
+ final RelDataType type = validator.deriveTypeImpl(scope, call);
+ final SqlOperator operator = call.getOperator();
+ final SqlCallBinding callBinding =
+ new SqlCallBinding(validator, scope, call);
+ // legacy table functions
+ if (operator instanceof TableSqlFunction) {
Review comment:
It seems that we can use `TableSqlFunction`, because `TableSqlFunction`s are
different between legacy planner and blink planner.
This lead to we have different codes in between legacy planner and blink
planner.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r378224696
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/validate/ProcedureNamespace.java
##
@@ -0,0 +1,80 @@
+/*
+ * 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.calcite.sql.validate;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.functions.utils.TableSqlFunction;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlUtil;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+/**
+ * Namespace whose contents are defined by the result of a call to a
user-defined procedure.
+ *
+ * Note: Compared to Calcite, this class implements custom logic for
dealing with collection tables
+ * like {@code TABLE(function(...))} procedures. Compared to the SQL standard,
Flink's table functions
Review comment:
OK, I got the meaning of "collection table".
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r378167489
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/validate/ProcedureNamespace.java
##
@@ -0,0 +1,80 @@
+/*
+ * 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.calcite.sql.validate;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.functions.utils.TableSqlFunction;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlUtil;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+/**
+ * Namespace whose contents are defined by the result of a call to a
user-defined procedure.
+ *
+ * Note: Compared to Calcite, this class implements custom logic for
dealing with collection tables
+ * like {@code TABLE(function(...))} procedures. Compared to the SQL standard,
Flink's table functions
Review comment:
But no input reference fields?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions URL: https://github.com/apache/flink/pull/11034#discussion_r376223974 ## File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/TableFunctionCallGen.scala ## @@ -49,7 +71,7 @@ class TableFunctionCallGen(tableFunction: TableFunction[_]) extends CallGenerato // has no result GeneratedExpression( - functionReference, + resultCollectorTerm, Review comment: A `resultCollectorTerm` instead of `functionReference` seems complicated. Can you add comments into `TableFunctionCallGen` to explain this. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r376220830
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/TableFunctionCallGen.scala
##
@@ -60,7 +82,61 @@ class TableFunctionCallGen(tableFunction: TableFunction[_])
extends CallGenerato
operands: Seq[GeneratedExpression],
func: TableFunction[_]): Array[GeneratedExpression] = {
// get the expanded parameter types
-var paramClasses = getEvalMethodSignature(func,
operands.map(_.resultType).toArray)
+val paramClasses = getEvalMethodSignature(func,
operands.map(_.resultType).toArray)
prepareFunctionArgs(ctx, operands, paramClasses,
func.getParameterTypes(paramClasses))
}
+
+ def getExternalDataType: DataType = {
+val sqlFunction = rexCall.getOperator.asInstanceOf[TableSqlFunction]
+val arguments =
UserDefinedFunctionUtils.transformRexNodes(rexCall.operands)
+val operandTypes = rexCall.operands
+.map(_.getType)
+.map(FlinkTypeFactory.toLogicalType).toArray
+val func = sqlFunction.makeFunction(arguments, operandTypes)
+val argTypes = getEvalMethodSignature(
+ func,
+ rexCall.operands
+.map(_.getType)
+.map(FlinkTypeFactory.toLogicalType).toArray)
+sqlFunction
+.getFunction
+.asInstanceOf[FlinkTableFunction]
+.getExternalResultType(func, arguments, argTypes)
+ }
+
+ /**
+ * Generates a collector that converts the output of a table function
(possibly as an atomic type)
+ * into an internal row type. Returns a collector term for referencing the
collector.
+ */
+ def generateResultCollector(ctx: CodeGeneratorContext): String = {
Review comment:
really like this refactoring: move pojo and external conversion things to
here, and this consistent with ScalarFunction too.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r376220459
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/TableFunctionCallGen.scala
##
@@ -18,29 +18,51 @@
package org.apache.flink.table.planner.codegen.calls
+import org.apache.calcite.rex.RexCall
+import org.apache.flink.table.dataformat.GenericRow
import org.apache.flink.table.functions.TableFunction
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
+import org.apache.flink.table.planner.codegen.CodeGenUtils.newName
import org.apache.flink.table.planner.codegen.GeneratedExpression.NEVER_NULL
import
org.apache.flink.table.planner.codegen.calls.ScalarFunctionCallGen.prepareFunctionArgs
-import org.apache.flink.table.planner.codegen.{CodeGeneratorContext,
GeneratedExpression}
+import org.apache.flink.table.planner.codegen._
import
org.apache.flink.table.planner.functions.utils.UserDefinedFunctionUtils.getEvalMethodSignature
+import org.apache.flink.table.planner.functions.utils.{TableSqlFunction,
UserDefinedFunctionUtils}
+import org.apache.flink.table.planner.plan.schema.FlinkTableFunction
+import
org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter.fromDataTypeToLogicalType
+import org.apache.flink.table.runtime.types.PlannerTypeUtils
+import org.apache.flink.table.types.DataType
import org.apache.flink.table.types.logical.LogicalType
+import scala.collection.JavaConversions._
+
/**
* Generates a call to user-defined [[TableFunction]].
*
* @param tableFunction user-defined [[TableFunction]] that might be
overloaded
*/
-class TableFunctionCallGen(tableFunction: TableFunction[_]) extends
CallGenerator {
+class TableFunctionCallGen(
+rexCall: RexCall,
+tableFunction: TableFunction[_])
+ extends CallGenerator {
override def generate(
ctx: CodeGeneratorContext,
operands: Seq[GeneratedExpression],
returnType: LogicalType): GeneratedExpression = {
-// convert parameters for function (output boxing)
-val parameters = prepareUDFArgs(ctx, operands, tableFunction)
-// generate function call
val functionReference = ctx.addReusableFunction(tableFunction)
+
+val resultCollectorTerm = generateResultCollector(ctx)
+
+val setCollectorCode = s"""
Review comment:
Minor: just `s"$functionReference.setCollector($resultCollectorTerm);"`?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r376224405
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/collector/WrappingCollector.java
##
@@ -0,0 +1,49 @@
+/*
+ * 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.runtime.collector;
+
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.util.Collector;
+
+/**
+ * A {@link Collector} that wraps another collector. An implementation can
decide when to emit to the
+ * wrapped collector.
+ */
+public abstract class WrappingCollector extends AbstractRichFunction
implements Collector {
+
+ private Collector collector;
+
+ /**
+* Sets the current collector which is used to emit the final result.
+*/
+ public void setCollector(Collector collector) {
+ this.collector = collector;
+ }
+
+ /**
+* Outputs the final result to the wrapped collector.
+*/
+ public void outputResult(T result) {
+ this.collector.collect(result);
+ }
+
+ public void close() {
Review comment:
`@Override`
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r376197098
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/validate/ProcedureNamespace.java
##
@@ -0,0 +1,80 @@
+/*
+ * 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.calcite.sql.validate;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.functions.utils.TableSqlFunction;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlUtil;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+/**
+ * Namespace whose contents are defined by the result of a call to a
user-defined procedure.
+ *
+ * Note: Compared to Calcite, this class implements custom logic for
dealing with collection tables
+ * like {@code TABLE(function(...))} procedures. Compared to the SQL standard,
Flink's table functions
Review comment:
I don't quite understand `TABLE(function(...))`, it seems only
`TABLE(function(literals))` support? Is this useful?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r376225561
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/TableFunctionCallGen.scala
##
@@ -60,7 +82,61 @@ class TableFunctionCallGen(tableFunction: TableFunction[_])
extends CallGenerato
operands: Seq[GeneratedExpression],
func: TableFunction[_]): Array[GeneratedExpression] = {
// get the expanded parameter types
-var paramClasses = getEvalMethodSignature(func,
operands.map(_.resultType).toArray)
+val paramClasses = getEvalMethodSignature(func,
operands.map(_.resultType).toArray)
prepareFunctionArgs(ctx, operands, paramClasses,
func.getParameterTypes(paramClasses))
}
+
+ def getExternalDataType: DataType = {
+val sqlFunction = rexCall.getOperator.asInstanceOf[TableSqlFunction]
+val arguments =
UserDefinedFunctionUtils.transformRexNodes(rexCall.operands)
+val operandTypes = rexCall.operands
+.map(_.getType)
+.map(FlinkTypeFactory.toLogicalType).toArray
+val func = sqlFunction.makeFunction(arguments, operandTypes)
+val argTypes = getEvalMethodSignature(
+ func,
+ rexCall.operands
+.map(_.getType)
+.map(FlinkTypeFactory.toLogicalType).toArray)
+sqlFunction
+.getFunction
+.asInstanceOf[FlinkTableFunction]
+.getExternalResultType(func, arguments, argTypes)
+ }
+
+ /**
+ * Generates a collector that converts the output of a table function
(possibly as an atomic type)
+ * into an internal row type. Returns a collector term for referencing the
collector.
+ */
+ def generateResultCollector(ctx: CodeGeneratorContext): String = {
+val externalDataType = getExternalDataType
+val pojoFieldMapping =
Some(UserDefinedFunctionUtils.getFieldInfo(externalDataType)._2)
+val externalType = fromDataTypeToLogicalType(externalDataType)
+val wrappedInternalType = PlannerTypeUtils.toRowType(externalType)
+
+val collectorCtx = CodeGeneratorContext(ctx.tableConfig)
+val externalTerm = newName("externalRecord")
+val resultGenerator = new ExprCodeGenerator(collectorCtx, true)
+ .bindInput(externalType, externalTerm, pojoFieldMapping)
+val wrappedResult = resultGenerator.generateConverterResultExpression(
Review comment:
It seems that we don't need this conversion when result type is not a pojo
and atomic object.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r376199644
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
##
@@ -295,10 +295,12 @@ public RelNode visit(SortQueryOperation sort) {
List parameters =
convertToRexNodes(calculatedTable.getParameters());
+ // TODO use relBuilder.functionScan() once we remove
TableSqlFunction
return LogicalTableFunctionScan.create(
relBuilder.peek().getCluster(),
Collections.emptyList(),
- relBuilder.call(sqlFunction,
parameters),
+ relBuilder.getRexBuilder()
+
.makeCall(function.getRowType(typeFactory, null, null), sqlFunction,
parameters),
Review comment:
Can you modify the `FlinkTableFunction`, just remove `arguments:
Array[AnyRef]` and `argTypes: Array[Class[_]`, them are useless. The follow-up
plan doesn't depend on them either.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11034: [FLINK-15802][table] Support new type inference for table functions
JingsongLi commented on a change in pull request #11034: [FLINK-15802][table]
Support new type inference for table functions
URL: https://github.com/apache/flink/pull/11034#discussion_r376226097
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/collector/WrappingCollector.java
##
@@ -0,0 +1,49 @@
+/*
+ * 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.runtime.collector;
+
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.util.Collector;
+
+/**
+ * A {@link Collector} that wraps another collector. An implementation can
decide when to emit to the
+ * wrapped collector.
+ */
+public abstract class WrappingCollector extends AbstractRichFunction
implements Collector {
+
+ private Collector collector;
+
+ /**
+* Sets the current collector which is used to emit the final result.
+*/
+ public void setCollector(Collector collector) {
+ this.collector = collector;
+ }
+
+ /**
+* Outputs the final result to the wrapped collector.
+*/
+ public void outputResult(T result) {
+ this.collector.collect(result);
Review comment:
Looks like this call will introduce a virtual method call to separate UDTF
invoking and subsequent projections.
Can we use `TableFunctionCollector` here?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
