You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2020/11/09 17:33:23 UTC
[GitHub] [iceberg] aokolnychyi opened a new pull request #1743: Spark: Add stored procedure API
aokolnychyi opened a new pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743
This PR adds the stored procedure API.
The actual implementation in Iceberg and tests will come in a separate PR.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r519994623
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, ident), args) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(ident)
+
+ validateParams(procedure)
+ validateMethodHandle(procedure)
+
+ Call(procedure, args = buildArgExprs(procedure, args))
+ }
+
+ private def validateParams(procedure: Procedure): Unit = {
+ // should not be any duplicate param names
+ val duplicateParamNames = procedure.parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(s"Duplicate parameter names: ${duplicateParamNames.mkString("[", ",", "]")}")
+ }
+
+ // optional params should be at the end
+ procedure.parameters.sliding(2).foreach {
+ case Array(previousParam, currentParam) if previousParam.required && !currentParam.required =>
+ throw new AnalysisException("Optional parameters must be after required ones")
+ case _ =>
+ }
+ }
+
+ private def validateMethodHandle(procedure: Procedure): Unit = {
+ val params = procedure.parameters
+ val outputType = procedure.outputType
+
+ val methodHandle = procedure.methodHandle
+ val methodType = methodHandle.`type`
+ val methodReturnType = methodType.returnType
+
+ // method cannot accept var ags
+ if (methodHandle.isVarargsCollector) {
+ throw new AnalysisException("Method must have fixed arity")
+ }
+
+ // verify the number of params in the procedure match the number of params in the method
+ if (params.length != methodType.parameterCount) {
+ throw new AnalysisException("Method parameter count must match the number of procedure parameters")
+ }
+
+ // the MethodHandle API does not allow us to check the generic type
+ // so we only verify the return type is either void or iterable
+
+ if (outputType.nonEmpty && methodReturnType != classOf[java.lang.Iterable[_]]) {
Review comment:
In the current approach, `methodHandle` is expected to be either void or return `java.lang.Iterable` of Spark Rows.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520065372
##########
File path: spark3-extensions/src/main/scala/org/apache/iceberg/spark/extensions/IcebergSparkSessionExtensions.scala
##########
@@ -20,13 +20,15 @@
package org.apache.iceberg.spark.extensions
import org.apache.spark.sql.SparkSessionExtensions
+import org.apache.spark.sql.catalyst.analysis.ResolveProcedures
import org.apache.spark.sql.catalyst.parser.extensions.IcebergSparkSqlExtensionsParser
import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Strategy
class IcebergSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
override def apply(extensions: SparkSessionExtensions): Unit = {
extensions.injectParser { case (_, parser) => new IcebergSparkSqlExtensionsParser(parser) }
+ extensions.injectResolutionRule { _ => ResolveProcedures }
Review comment:
The ignored argument is a `SparkSession`, which I think should be passed into `ResolveProcedures` so that it doesn't need to reference `SparkSession.active`.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520955021
##########
File path: project/scalastyle_config.xml
##########
@@ -85,11 +85,6 @@
<parameter name="maxTypes"><![CDATA[30]]></parameter>
</parameters>
</check>
- <check level="error" class="org.scalastyle.scalariform.CyclomaticComplexityChecker" enabled="true">
Review comment:
I can move this into a separate PR and disable in Java 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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520952116
##########
File path: project/scalastyle_config.xml
##########
@@ -85,11 +85,6 @@
<parameter name="maxTypes"><![CDATA[30]]></parameter>
</parameters>
</check>
- <check level="error" class="org.scalastyle.scalariform.CyclomaticComplexityChecker" enabled="true">
Review comment:
We've discussed how handy this rule is multiple times. I'd remove it if there are no objections.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520953283
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CallExec.scala
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.spark.sql.execution.datasources.v2
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.Procedure
+
+case class CallExec(
+ output: Seq[Attribute],
+ procedure: Procedure,
+ input: InternalRow) extends V2CommandExec {
+
+ override protected def run(): Seq[InternalRow] = {
+ procedure.call(input)
Review comment:
@rdblue, I do not pass a type alongside the row since the rules rely on `parameters` reported by the procedure to align input expressions.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520163076
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, ident), args) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(ident)
+
+ validateParams(procedure)
+ validateMethodHandle(procedure)
+
+ Call(procedure, args = buildArgExprs(procedure, args))
+ }
+
+ private def validateParams(procedure: Procedure): Unit = {
+ // should not be any duplicate param names
+ val duplicateParamNames = procedure.parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(s"Duplicate parameter names: ${duplicateParamNames.mkString("[", ",", "]")}")
+ }
+
+ // optional params should be at the end
+ procedure.parameters.sliding(2).foreach {
+ case Array(previousParam, currentParam) if previousParam.required && !currentParam.required =>
+ throw new AnalysisException("Optional parameters must be after required ones")
+ case _ =>
+ }
+ }
+
+ private def validateMethodHandle(procedure: Procedure): Unit = {
+ val params = procedure.parameters
+ val outputType = procedure.outputType
+
+ val methodHandle = procedure.methodHandle
+ val methodType = methodHandle.`type`
+ val methodReturnType = methodType.returnType
+
+ // method cannot accept var ags
+ if (methodHandle.isVarargsCollector) {
+ throw new AnalysisException("Method must have fixed arity")
+ }
+
+ // verify the number of params in the procedure match the number of params in the method
+ if (params.length != methodType.parameterCount) {
+ throw new AnalysisException("Method parameter count must match the number of procedure parameters")
+ }
+
+ // the MethodHandle API does not allow us to check the generic type
+ // so we only verify the return type is either void or iterable
+
+ if (outputType.nonEmpty && methodReturnType != classOf[java.lang.Iterable[_]]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines $outputType " +
+ "as its output so must return java.lang.Iterable of Spark Rows")
+ }
+
+ if (outputType.isEmpty && methodReturnType != classOf[Void]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines no output columns " +
+ "so must be void")
+ }
+ }
+
+ private def buildArgExprs(procedure: Procedure, args: Seq[CallArgument]): Seq[Expression] = {
+ val params = procedure.parameters
+
+ // build a map of declared parameter names to their positions
+ val nameToPositionMap = params.map(_.name).zipWithIndex.toMap
+
+ // build a map of parameter names to args
+ val nameToArgMap = buildNameToArgMap(params, args, nameToPositionMap)
+
+ // verify all required parameters are provided
+ val missingParamNames = params.filter(_.required).collect {
+ case param if !nameToArgMap.contains(param.name) => param.name
+ }
+
+ if (missingParamNames.nonEmpty) {
+ throw new AnalysisException(s"Missing required parameters: ${missingParamNames.mkString("[", ",", "]")}")
+ }
+
+ val argExprs = new Array[Expression](params.size)
+
+ nameToArgMap.foreach { case (name, arg) =>
+ val position = nameToPositionMap(name)
+ val param = params(position)
+ val paramType = param.dataType
+ val argType = arg.expr.dataType
+ if (paramType != argType) {
+ throw new AnalysisException(s"Wrong arg type for ${param.name}: expected $paramType but got $argType")
Review comment:
This is going to force people to use value literals, like `12L`. Can we insert a cast and execute it if the type is compatible and the cast is safe? I'm thinking it would be fine for int -> long, but not for string -> int.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r521481156
##########
File path: spark3/src/main/java/org/apache/spark/sql/connector/catalog/ProcedureParameter.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.spark.sql.connector.catalog;
+
+import org.apache.spark.sql.types.DataType;
+
+public interface ProcedureParameter {
Review comment:
I'll add Javadoc in a follow-up PR. I need to document other places like `IcebergSparkSessionExtensions`. I want to have at least one procedure to be merged using these APIs.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520131868
##########
File path: spark3/src/main/java/org/apache/spark/sql/connector/catalog/Procedure.java
##########
@@ -0,0 +1,29 @@
+/*
+ * 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.spark.sql.connector.catalog;
+
+import java.lang.invoke.MethodHandle;
+import org.apache.spark.sql.types.StructType;
+
+public interface Procedure {
+ ProcedureParameter[] parameters();
+ StructType outputType();
Review comment:
Does this need to be a struct?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520065571
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
Review comment:
The session should be passed in from rule injection.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520066422
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ExtendedDataSourceV2Strategy.scala
##########
@@ -20,14 +20,21 @@
package org.apache.spark.sql.execution.datasources.v2
import org.apache.spark.sql.{AnalysisException, Strategy}
-import org.apache.spark.sql.catalyst.plans.logical.{CallStatement, LogicalPlan}
+import org.apache.spark.sql.catalyst.CatalystTypeConverters
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, LogicalPlan}
import org.apache.spark.sql.execution.SparkPlan
object ExtendedDataSourceV2Strategy extends Strategy {
override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
- case _: CallStatement =>
- throw new AnalysisException("CALL statements are not currently supported")
+ case c @ Call(procedure, args) =>
+ CallExec(c.output, procedure.methodHandle, args.map(toScalaValue)) :: Nil
Review comment:
Should this be in `ExtendedDataSourceV2Strategy`? It doesn't seem related to DSv2.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r521479793
##########
File path: spark3/src/main/java/org/apache/spark/sql/connector/catalog/Procedure.java
##########
@@ -0,0 +1,29 @@
+/*
+ * 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.spark.sql.connector.catalog;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.StructType;
+
+public interface Procedure {
Review comment:
Done.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520952581
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, ident), args) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(ident)
+
+ validateParams(procedure)
+ validateMethodHandle(procedure)
+
+ Call(procedure, args = buildArgExprs(procedure, args))
+ }
+
+ private def validateParams(procedure: Procedure): Unit = {
+ // should not be any duplicate param names
+ val duplicateParamNames = procedure.parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(s"Duplicate parameter names: ${duplicateParamNames.mkString("[", ",", "]")}")
+ }
+
+ // optional params should be at the end
+ procedure.parameters.sliding(2).foreach {
+ case Array(previousParam, currentParam) if previousParam.required && !currentParam.required =>
+ throw new AnalysisException("Optional parameters must be after required ones")
+ case _ =>
+ }
+ }
+
+ private def validateMethodHandle(procedure: Procedure): Unit = {
+ val params = procedure.parameters
+ val outputType = procedure.outputType
+
+ val methodHandle = procedure.methodHandle
+ val methodType = methodHandle.`type`
+ val methodReturnType = methodType.returnType
+
+ // method cannot accept var ags
+ if (methodHandle.isVarargsCollector) {
+ throw new AnalysisException("Method must have fixed arity")
+ }
+
+ // verify the number of params in the procedure match the number of params in the method
+ if (params.length != methodType.parameterCount) {
+ throw new AnalysisException("Method parameter count must match the number of procedure parameters")
+ }
+
+ // the MethodHandle API does not allow us to check the generic type
+ // so we only verify the return type is either void or iterable
+
+ if (outputType.nonEmpty && methodReturnType != classOf[java.lang.Iterable[_]]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines $outputType " +
+ "as its output so must return java.lang.Iterable of Spark Rows")
+ }
+
+ if (outputType.isEmpty && methodReturnType != classOf[Void]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines no output columns " +
+ "so must be void")
+ }
+ }
+
+ private def buildArgExprs(procedure: Procedure, args: Seq[CallArgument]): Seq[Expression] = {
+ val params = procedure.parameters
+
+ // build a map of declared parameter names to their positions
+ val nameToPositionMap = params.map(_.name).zipWithIndex.toMap
+
+ // build a map of parameter names to args
+ val nameToArgMap = buildNameToArgMap(params, args, nameToPositionMap)
+
+ // verify all required parameters are provided
+ val missingParamNames = params.filter(_.required).collect {
+ case param if !nameToArgMap.contains(param.name) => param.name
+ }
+
+ if (missingParamNames.nonEmpty) {
+ throw new AnalysisException(s"Missing required parameters: ${missingParamNames.mkString("[", ",", "]")}")
+ }
+
+ val argExprs = new Array[Expression](params.size)
+
+ nameToArgMap.foreach { case (name, arg) =>
+ val position = nameToPositionMap(name)
+ val param = params(position)
+ val paramType = param.dataType
+ val argType = arg.expr.dataType
+ if (paramType != argType) {
+ throw new AnalysisException(s"Wrong arg type for ${param.name}: expected $paramType but got $argType")
Review comment:
All expressions will undergo optimizations so we will get rid of casts before creating an exec node.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520163819
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, ident), args) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(ident)
+
+ validateParams(procedure)
+ validateMethodHandle(procedure)
+
+ Call(procedure, args = buildArgExprs(procedure, args))
+ }
+
+ private def validateParams(procedure: Procedure): Unit = {
+ // should not be any duplicate param names
+ val duplicateParamNames = procedure.parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(s"Duplicate parameter names: ${duplicateParamNames.mkString("[", ",", "]")}")
+ }
+
+ // optional params should be at the end
+ procedure.parameters.sliding(2).foreach {
+ case Array(previousParam, currentParam) if previousParam.required && !currentParam.required =>
+ throw new AnalysisException("Optional parameters must be after required ones")
+ case _ =>
+ }
+ }
+
+ private def validateMethodHandle(procedure: Procedure): Unit = {
+ val params = procedure.parameters
+ val outputType = procedure.outputType
+
+ val methodHandle = procedure.methodHandle
+ val methodType = methodHandle.`type`
+ val methodReturnType = methodType.returnType
+
+ // method cannot accept var ags
+ if (methodHandle.isVarargsCollector) {
+ throw new AnalysisException("Method must have fixed arity")
+ }
+
+ // verify the number of params in the procedure match the number of params in the method
+ if (params.length != methodType.parameterCount) {
+ throw new AnalysisException("Method parameter count must match the number of procedure parameters")
+ }
+
+ // the MethodHandle API does not allow us to check the generic type
+ // so we only verify the return type is either void or iterable
+
+ if (outputType.nonEmpty && methodReturnType != classOf[java.lang.Iterable[_]]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines $outputType " +
+ "as its output so must return java.lang.Iterable of Spark Rows")
+ }
+
+ if (outputType.isEmpty && methodReturnType != classOf[Void]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines no output columns " +
+ "so must be void")
+ }
+ }
+
+ private def buildArgExprs(procedure: Procedure, args: Seq[CallArgument]): Seq[Expression] = {
+ val params = procedure.parameters
+
+ // build a map of declared parameter names to their positions
+ val nameToPositionMap = params.map(_.name).zipWithIndex.toMap
+
+ // build a map of parameter names to args
+ val nameToArgMap = buildNameToArgMap(params, args, nameToPositionMap)
+
+ // verify all required parameters are provided
+ val missingParamNames = params.filter(_.required).collect {
+ case param if !nameToArgMap.contains(param.name) => param.name
+ }
+
+ if (missingParamNames.nonEmpty) {
+ throw new AnalysisException(s"Missing required parameters: ${missingParamNames.mkString("[", ",", "]")}")
+ }
+
+ val argExprs = new Array[Expression](params.size)
+
+ nameToArgMap.foreach { case (name, arg) =>
+ val position = nameToPositionMap(name)
+ val param = params(position)
+ val paramType = param.dataType
+ val argType = arg.expr.dataType
+ if (paramType != argType) {
+ throw new AnalysisException(s"Wrong arg type for ${param.name}: expected $paramType but got $argType")
+ }
+ argExprs(position) = arg.expr
+ }
+
+ // assign nulls to optional params that were not set
+ params.foreach {
+ case p if !p.required && !nameToArgMap.contains(p.name) =>
+ val position = nameToPositionMap(p.name)
+ argExprs(position) = Literal.create(null, p.dataType)
+ case _ =>
+ }
+
+ argExprs
+ }
+
+ private def buildNameToArgMap(
+ params: Seq[ProcedureParameter],
+ args: Seq[CallArgument],
+ nameToPositionMap: Map[String, Int]): Map[String, CallArgument] = {
+
+ val containsNamedArg = args.exists(_.isInstanceOf[NamedArgument])
+ val containsPositionalArg = args.exists(_.isInstanceOf[PositionalArgument])
+
+ if (containsNamedArg && containsPositionalArg) {
+ throw new AnalysisException("Named and positional arguments cannot be mixed")
Review comment:
This is Presto's current behavior?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520953862
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, ident), args) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(ident)
+
+ validateParams(procedure)
+ validateMethodHandle(procedure)
+
+ Call(procedure, args = buildArgExprs(procedure, args))
+ }
+
+ private def validateParams(procedure: Procedure): Unit = {
+ // should not be any duplicate param names
+ val duplicateParamNames = procedure.parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(s"Duplicate parameter names: ${duplicateParamNames.mkString("[", ",", "]")}")
+ }
+
+ // optional params should be at the end
+ procedure.parameters.sliding(2).foreach {
+ case Array(previousParam, currentParam) if previousParam.required && !currentParam.required =>
+ throw new AnalysisException("Optional parameters must be after required ones")
+ case _ =>
+ }
+ }
+
+ private def validateMethodHandle(procedure: Procedure): Unit = {
+ val params = procedure.parameters
+ val outputType = procedure.outputType
+
+ val methodHandle = procedure.methodHandle
+ val methodType = methodHandle.`type`
+ val methodReturnType = methodType.returnType
+
+ // method cannot accept var ags
+ if (methodHandle.isVarargsCollector) {
+ throw new AnalysisException("Method must have fixed arity")
+ }
+
+ // verify the number of params in the procedure match the number of params in the method
+ if (params.length != methodType.parameterCount) {
+ throw new AnalysisException("Method parameter count must match the number of procedure parameters")
+ }
+
+ // the MethodHandle API does not allow us to check the generic type
+ // so we only verify the return type is either void or iterable
+
+ if (outputType.nonEmpty && methodReturnType != classOf[java.lang.Iterable[_]]) {
Review comment:
Got rid of this.
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
Review comment:
Done.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520954155
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ExtendedDataSourceV2Strategy.scala
##########
@@ -20,14 +20,21 @@
package org.apache.spark.sql.execution.datasources.v2
import org.apache.spark.sql.{AnalysisException, Strategy}
-import org.apache.spark.sql.catalyst.plans.logical.{CallStatement, LogicalPlan}
+import org.apache.spark.sql.catalyst.CatalystTypeConverters
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, LogicalPlan}
import org.apache.spark.sql.execution.SparkPlan
object ExtendedDataSourceV2Strategy extends Strategy {
override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
- case _: CallStatement =>
- throw new AnalysisException("CALL statements are not currently supported")
+ case c @ Call(procedure, args) =>
+ CallExec(c.output, procedure.methodHandle, args.map(toScalaValue)) :: Nil
Review comment:
@rdblue, I think this is debatable. It does depend on the catalog API part of DS V2 so I've put it here. Any suggestions?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue merged pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue merged pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520968257
##########
File path: spark3/src/main/java/org/apache/spark/sql/connector/catalog/ProcedureParameter.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.spark.sql.connector.catalog;
+
+import org.apache.spark.sql.types.DataType;
+
+public interface ProcedureParameter {
Review comment:
I think for all of the connector interfaces, we should add clear Javadoc.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520921190
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, ident), args) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(ident)
+
+ validateParams(procedure)
+ validateMethodHandle(procedure)
+
+ Call(procedure, args = buildArgExprs(procedure, args))
+ }
+
+ private def validateParams(procedure: Procedure): Unit = {
+ // should not be any duplicate param names
+ val duplicateParamNames = procedure.parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(s"Duplicate parameter names: ${duplicateParamNames.mkString("[", ",", "]")}")
+ }
+
+ // optional params should be at the end
+ procedure.parameters.sliding(2).foreach {
+ case Array(previousParam, currentParam) if previousParam.required && !currentParam.required =>
+ throw new AnalysisException("Optional parameters must be after required ones")
+ case _ =>
+ }
+ }
+
+ private def validateMethodHandle(procedure: Procedure): Unit = {
+ val params = procedure.parameters
+ val outputType = procedure.outputType
+
+ val methodHandle = procedure.methodHandle
+ val methodType = methodHandle.`type`
+ val methodReturnType = methodType.returnType
+
+ // method cannot accept var ags
+ if (methodHandle.isVarargsCollector) {
+ throw new AnalysisException("Method must have fixed arity")
+ }
+
+ // verify the number of params in the procedure match the number of params in the method
+ if (params.length != methodType.parameterCount) {
+ throw new AnalysisException("Method parameter count must match the number of procedure parameters")
+ }
+
+ // the MethodHandle API does not allow us to check the generic type
+ // so we only verify the return type is either void or iterable
+
+ if (outputType.nonEmpty && methodReturnType != classOf[java.lang.Iterable[_]]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines $outputType " +
+ "as its output so must return java.lang.Iterable of Spark Rows")
+ }
+
+ if (outputType.isEmpty && methodReturnType != classOf[Void]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines no output columns " +
+ "so must be void")
+ }
+ }
+
+ private def buildArgExprs(procedure: Procedure, args: Seq[CallArgument]): Seq[Expression] = {
+ val params = procedure.parameters
+
+ // build a map of declared parameter names to their positions
+ val nameToPositionMap = params.map(_.name).zipWithIndex.toMap
+
+ // build a map of parameter names to args
+ val nameToArgMap = buildNameToArgMap(params, args, nameToPositionMap)
+
+ // verify all required parameters are provided
+ val missingParamNames = params.filter(_.required).collect {
+ case param if !nameToArgMap.contains(param.name) => param.name
+ }
+
+ if (missingParamNames.nonEmpty) {
+ throw new AnalysisException(s"Missing required parameters: ${missingParamNames.mkString("[", ",", "]")}")
+ }
+
+ val argExprs = new Array[Expression](params.size)
+
+ nameToArgMap.foreach { case (name, arg) =>
+ val position = nameToPositionMap(name)
+ val param = params(position)
+ val paramType = param.dataType
+ val argType = arg.expr.dataType
+ if (paramType != argType) {
+ throw new AnalysisException(s"Wrong arg type for ${param.name}: expected $paramType but got $argType")
+ }
+ argExprs(position) = arg.expr
+ }
+
+ // assign nulls to optional params that were not set
+ params.foreach {
+ case p if !p.required && !nameToArgMap.contains(p.name) =>
+ val position = nameToPositionMap(p.name)
+ argExprs(position) = Literal.create(null, p.dataType)
+ case _ =>
+ }
+
+ argExprs
+ }
+
+ private def buildNameToArgMap(
+ params: Seq[ProcedureParameter],
+ args: Seq[CallArgument],
+ nameToPositionMap: Map[String, Int]): Map[String, CallArgument] = {
+
+ val containsNamedArg = args.exists(_.isInstanceOf[NamedArgument])
+ val containsPositionalArg = args.exists(_.isInstanceOf[PositionalArgument])
+
+ if (containsNamedArg && containsPositionalArg) {
+ throw new AnalysisException("Named and positional arguments cannot be mixed")
Review comment:
Yes, Presto does not allow mixing named and positional args.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520953595
##########
File path: spark3/src/main/java/org/apache/spark/sql/connector/catalog/Procedure.java
##########
@@ -0,0 +1,29 @@
+/*
+ * 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.spark.sql.connector.catalog;
+
+import java.lang.invoke.MethodHandle;
+import org.apache.spark.sql.types.StructType;
+
+public interface Procedure {
+ ProcedureParameter[] parameters();
+ StructType outputType();
Review comment:
Procedures are supposed to return Spark rows that will be reported as output. So it is the type of those rows.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520967197
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/Call.scala
##########
@@ -0,0 +1,28 @@
+/*
+ * 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.spark.sql.catalyst.plans.logical
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.connector.catalog.Procedure
+import scala.collection.Seq
+
+case class Call(procedure: Procedure, args: Seq[Expression]) extends Command {
+ override def output: Seq[Attribute] = procedure.outputType.toAttributes
Review comment:
I think this needs to be a lazy val, or else the attributes will have different IDs every time it is called.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r521479415
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/Call.scala
##########
@@ -0,0 +1,33 @@
+/*
+ * 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.spark.sql.catalyst.plans.logical
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.catalyst.util.truncatedString
+import org.apache.spark.sql.connector.catalog.Procedure
+import scala.collection.Seq
+
+case class Call(procedure: Procedure, args: Seq[Expression]) extends Command {
+ override lazy val output: Seq[Attribute] = procedure.outputType.toAttributes
+
+ override def simpleString(maxFields: Int): String = {
Review comment:
@rdblue, I implemented `simpleString` in `Call` and `CallExec` that use `description` in `Procedure`.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520969875
##########
File path: spark3/src/main/java/org/apache/spark/sql/connector/catalog/Procedure.java
##########
@@ -0,0 +1,29 @@
+/*
+ * 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.spark.sql.connector.catalog;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.StructType;
+
+public interface Procedure {
Review comment:
One more thing: we should also add a `describe` method so that we can show these correctly in plans.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520966189
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Cast, Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+case class ResolveProcedures(spark: SparkSession) extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = spark.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, ident), args) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(ident)
+ validateParams(procedure)
+ Call(procedure, args = buildArgExprs(procedure, args))
+ }
+
+ private def validateParams(procedure: Procedure): Unit = {
+ // should not be any duplicate param names
+ val duplicateParamNames = procedure.parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(s"Duplicate parameter names: ${duplicateParamNames.mkString("[", ",", "]")}")
+ }
+
+ // optional params should be at the end
+ procedure.parameters.sliding(2).foreach {
+ case Array(previousParam, currentParam) if previousParam.required && !currentParam.required =>
+ throw new AnalysisException("Optional parameters must be after required ones")
+ case _ =>
+ }
+ }
+
+ private def buildArgExprs(procedure: Procedure, args: Seq[CallArgument]): Seq[Expression] = {
+ val params = procedure.parameters
+
+ // build a map of declared parameter names to their positions
+ val nameToPositionMap = params.map(_.name).zipWithIndex.toMap
+
+ // build a map of parameter names to args
+ val nameToArgMap = buildNameToArgMap(params, args, nameToPositionMap)
+
+ // verify all required parameters are provided
+ val missingParamNames = params.filter(_.required).collect {
+ case param if !nameToArgMap.contains(param.name) => param.name
+ }
+
+ if (missingParamNames.nonEmpty) {
+ throw new AnalysisException(s"Missing required parameters: ${missingParamNames.mkString("[", ",", "]")}")
+ }
+
+ val argExprs = new Array[Expression](params.size)
+
+ nameToArgMap.foreach { case (name, arg) =>
+ val position = nameToPositionMap(name)
+ val param = params(position)
+ val paramType = param.dataType
+ val argType = arg.expr.dataType
+
+ if (paramType != argType && !Cast.canUpCast(argType, paramType)) {
+ throw new AnalysisException(s"Wrong arg type for ${param.name}: expected $paramType but got $argType")
+ }
+
+ if (paramType != argType) {
+ argExprs(position) = Cast(arg.expr, paramType)
+ } else {
+ argExprs(position) = arg.expr
+ }
+ }
+
+ // assign nulls to optional params that were not set
+ params.foreach {
+ case p if !p.required && !nameToArgMap.contains(p.name) =>
Review comment:
We know from validation above that if the map doesn't contain an argument, that it must be required. Not a big deal to have the additional check, though.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#issuecomment-724351856
Mostly looks good, but I would probably pass arguments using an interface that passes InternalRow rather than using MethodHandle.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#issuecomment-725598764
Looks great. I'm committing this so we can start building stored procedures for actions!
Thanks for getting this done!
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520952298
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, ident), args) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(ident)
+
+ validateParams(procedure)
+ validateMethodHandle(procedure)
+
+ Call(procedure, args = buildArgExprs(procedure, args))
+ }
+
+ private def validateParams(procedure: Procedure): Unit = {
+ // should not be any duplicate param names
+ val duplicateParamNames = procedure.parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(s"Duplicate parameter names: ${duplicateParamNames.mkString("[", ",", "]")}")
+ }
+
+ // optional params should be at the end
+ procedure.parameters.sliding(2).foreach {
+ case Array(previousParam, currentParam) if previousParam.required && !currentParam.required =>
+ throw new AnalysisException("Optional parameters must be after required ones")
+ case _ =>
+ }
+ }
+
+ private def validateMethodHandle(procedure: Procedure): Unit = {
+ val params = procedure.parameters
+ val outputType = procedure.outputType
+
+ val methodHandle = procedure.methodHandle
+ val methodType = methodHandle.`type`
+ val methodReturnType = methodType.returnType
+
+ // method cannot accept var ags
+ if (methodHandle.isVarargsCollector) {
+ throw new AnalysisException("Method must have fixed arity")
+ }
+
+ // verify the number of params in the procedure match the number of params in the method
+ if (params.length != methodType.parameterCount) {
+ throw new AnalysisException("Method parameter count must match the number of procedure parameters")
+ }
+
+ // the MethodHandle API does not allow us to check the generic type
+ // so we only verify the return type is either void or iterable
+
+ if (outputType.nonEmpty && methodReturnType != classOf[java.lang.Iterable[_]]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines $outputType " +
+ "as its output so must return java.lang.Iterable of Spark Rows")
+ }
+
+ if (outputType.isEmpty && methodReturnType != classOf[Void]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines no output columns " +
+ "so must be void")
+ }
+ }
+
+ private def buildArgExprs(procedure: Procedure, args: Seq[CallArgument]): Seq[Expression] = {
+ val params = procedure.parameters
+
+ // build a map of declared parameter names to their positions
+ val nameToPositionMap = params.map(_.name).zipWithIndex.toMap
+
+ // build a map of parameter names to args
+ val nameToArgMap = buildNameToArgMap(params, args, nameToPositionMap)
+
+ // verify all required parameters are provided
+ val missingParamNames = params.filter(_.required).collect {
+ case param if !nameToArgMap.contains(param.name) => param.name
+ }
+
+ if (missingParamNames.nonEmpty) {
+ throw new AnalysisException(s"Missing required parameters: ${missingParamNames.mkString("[", ",", "]")}")
+ }
+
+ val argExprs = new Array[Expression](params.size)
+
+ nameToArgMap.foreach { case (name, arg) =>
+ val position = nameToPositionMap(name)
+ val param = params(position)
+ val paramType = param.dataType
+ val argType = arg.expr.dataType
+ if (paramType != argType) {
+ throw new AnalysisException(s"Wrong arg type for ${param.name}: expected $paramType but got $argType")
Review comment:
I added a cast if it is safe.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r521481690
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/Call.scala
##########
@@ -0,0 +1,28 @@
+/*
+ * 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.spark.sql.catalyst.plans.logical
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.connector.catalog.Procedure
+import scala.collection.Seq
+
+case class Call(procedure: Procedure, args: Seq[Expression]) extends Command {
+ override def output: Seq[Attribute] = procedure.outputType.toAttributes
Review comment:
Fixed.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520953774
##########
File path: spark3-extensions/src/main/scala/org/apache/iceberg/spark/extensions/IcebergSparkSessionExtensions.scala
##########
@@ -20,13 +20,15 @@
package org.apache.iceberg.spark.extensions
import org.apache.spark.sql.SparkSessionExtensions
+import org.apache.spark.sql.catalyst.analysis.ResolveProcedures
import org.apache.spark.sql.catalyst.parser.extensions.IcebergSparkSqlExtensionsParser
import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Strategy
class IcebergSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
override def apply(extensions: SparkSessionExtensions): Unit = {
extensions.injectParser { case (_, parser) => new IcebergSparkSqlExtensionsParser(parser) }
+ extensions.injectResolutionRule { _ => ResolveProcedures }
Review comment:
Done.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r520163076
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, ident), args) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(ident)
+
+ validateParams(procedure)
+ validateMethodHandle(procedure)
+
+ Call(procedure, args = buildArgExprs(procedure, args))
+ }
+
+ private def validateParams(procedure: Procedure): Unit = {
+ // should not be any duplicate param names
+ val duplicateParamNames = procedure.parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(s"Duplicate parameter names: ${duplicateParamNames.mkString("[", ",", "]")}")
+ }
+
+ // optional params should be at the end
+ procedure.parameters.sliding(2).foreach {
+ case Array(previousParam, currentParam) if previousParam.required && !currentParam.required =>
+ throw new AnalysisException("Optional parameters must be after required ones")
+ case _ =>
+ }
+ }
+
+ private def validateMethodHandle(procedure: Procedure): Unit = {
+ val params = procedure.parameters
+ val outputType = procedure.outputType
+
+ val methodHandle = procedure.methodHandle
+ val methodType = methodHandle.`type`
+ val methodReturnType = methodType.returnType
+
+ // method cannot accept var ags
+ if (methodHandle.isVarargsCollector) {
+ throw new AnalysisException("Method must have fixed arity")
+ }
+
+ // verify the number of params in the procedure match the number of params in the method
+ if (params.length != methodType.parameterCount) {
+ throw new AnalysisException("Method parameter count must match the number of procedure parameters")
+ }
+
+ // the MethodHandle API does not allow us to check the generic type
+ // so we only verify the return type is either void or iterable
+
+ if (outputType.nonEmpty && methodReturnType != classOf[java.lang.Iterable[_]]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines $outputType " +
+ "as its output so must return java.lang.Iterable of Spark Rows")
+ }
+
+ if (outputType.isEmpty && methodReturnType != classOf[Void]) {
+ throw new AnalysisException(
+ s"Wrong method return type: $methodReturnType; the procedure defines no output columns " +
+ "so must be void")
+ }
+ }
+
+ private def buildArgExprs(procedure: Procedure, args: Seq[CallArgument]): Seq[Expression] = {
+ val params = procedure.parameters
+
+ // build a map of declared parameter names to their positions
+ val nameToPositionMap = params.map(_.name).zipWithIndex.toMap
+
+ // build a map of parameter names to args
+ val nameToArgMap = buildNameToArgMap(params, args, nameToPositionMap)
+
+ // verify all required parameters are provided
+ val missingParamNames = params.filter(_.required).collect {
+ case param if !nameToArgMap.contains(param.name) => param.name
+ }
+
+ if (missingParamNames.nonEmpty) {
+ throw new AnalysisException(s"Missing required parameters: ${missingParamNames.mkString("[", ",", "]")}")
+ }
+
+ val argExprs = new Array[Expression](params.size)
+
+ nameToArgMap.foreach { case (name, arg) =>
+ val position = nameToPositionMap(name)
+ val param = params(position)
+ val paramType = param.dataType
+ val argType = arg.expr.dataType
+ if (paramType != argType) {
+ throw new AnalysisException(s"Wrong arg type for ${param.name}: expected $paramType but got $argType")
Review comment:
This is going to force people to use value literals, like `12L`. Can we insert a cast and execute it if the type is compatible?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r519998378
##########
File path: spark3/src/main/java/org/apache/spark/sql/connector/catalog/Procedure.java
##########
@@ -0,0 +1,29 @@
+/*
+ * 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.spark.sql.connector.catalog;
+
+import java.lang.invoke.MethodHandle;
+import org.apache.spark.sql.types.StructType;
+
+public interface Procedure {
+ ProcedureParameter[] parameters();
+ StructType outputType();
+ MethodHandle methodHandle();
Review comment:
Presto uses method handles for stored procedures 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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1743: Spark: Add stored procedure API
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1743:
URL: https://github.com/apache/iceberg/pull/1743#discussion_r519993784
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Call, CallArgument, CallStatement, LogicalPlan, NamedArgument, PositionalArgument}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, Procedure, ProcedureCatalog, ProcedureParameter}
+import scala.collection.Seq
+
+object ResolveProcedures extends Rule[LogicalPlan] with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager = SparkSession.active.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, ident), args) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(ident)
+
+ validateParams(procedure)
+ validateMethodHandle(procedure)
+
+ Call(procedure, args = buildArgExprs(procedure, args))
+ }
+
+ private def validateParams(procedure: Procedure): Unit = {
+ // should not be any duplicate param names
+ val duplicateParamNames = procedure.parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(s"Duplicate parameter names: ${duplicateParamNames.mkString("[", ",", "]")}")
+ }
+
+ // optional params should be at the end
+ procedure.parameters.sliding(2).foreach {
+ case Array(previousParam, currentParam) if previousParam.required && !currentParam.required =>
+ throw new AnalysisException("Optional parameters must be after required ones")
+ case _ =>
+ }
+ }
+
+ private def validateMethodHandle(procedure: Procedure): Unit = {
+ val params = procedure.parameters
+ val outputType = procedure.outputType
+
+ val methodHandle = procedure.methodHandle
+ val methodType = methodHandle.`type`
+ val methodReturnType = methodType.returnType
+
+ // method cannot accept var ags
+ if (methodHandle.isVarargsCollector) {
+ throw new AnalysisException("Method must have fixed arity")
+ }
+
+ // verify the number of params in the procedure match the number of params in the method
+ if (params.length != methodType.parameterCount) {
+ throw new AnalysisException("Method parameter count must match the number of procedure parameters")
+ }
+
+ // the MethodHandle API does not allow us to check the generic type
+ // so we only verify the return type is either void or iterable
+
+ if (outputType.nonEmpty && methodReturnType != classOf[java.lang.Iterable[_]]) {
Review comment:
This is ugly. I'd appreciate any ideas.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org