You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2022/08/10 02:30:00 UTC
[GitHub] [hudi] alexeykudinkin opened a new pull request, #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
alexeykudinkin opened a new pull request, #6352:
URL: https://github.com/apache/hudi/pull/6352
### Change Logs
`SQLConf` by default isn't propagated to Executors from the Driver. This is the reason why configuration specified w/ `--conf` is not being respected by some components being invoked on the executor (see https://github.com/apache/hudi/issues/6278)
In Spark this has been remediated by the introduction of `SQLExecutionRDD` that does snapshot `SQLConf` subsequently overriding it on the Executor side. However this has following limitations:
1. `SQLExecutionRDD` does propagate `SQLConf` only for RDDs that are **preceding** it (ie its dependencies), and doesn't for the ones that are chained after it.
2. `SQLExecutionRDD` is only applicable to `RDD[InternalRow]`
To work around #2 we're introduced `SQLConfInjectingRDD` (which is just a generalization of `SQLExecutionRDD`) and make sure that it's properly injected (to address #1) inside, for ex, `HoodieSparkUtils.createRdd` invocation (that is being carried out on the Executor side)
### Impact
None
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Change Logs and Impact were stated clearly
- [ ] Adequate tests were added if applicable
- [ ] CI passed
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #6352:
URL: https://github.com/apache/hudi/pull/6352#issuecomment-1210103122
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10715",
"triggerID" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10715)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua commented on a diff in pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
yihua commented on code in PR #6352:
URL: https://github.com/apache/hudi/pull/6352#discussion_r953160525
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/SQLConfInjectingRDD.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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
+
+import org.apache.spark.{Partition, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.internal.SQLConf
+
+import scala.reflect.ClassTag
+
+/**
+ * NOTE: This is a generalized version of of Spark's [[SQLExecutionRDD]]
+ *
+ * It is just a wrapper over [[sqlRDD]] which sets and makes effective all the configs from the
+ * captured [[SQLConf]]
+ *
+ * @param sqlRDD the `RDD` generated by the SQL plan
+ * @param conf the `SQLConf` to apply to the execution of the SQL plan
+ */
+class SQLConfInjectingRDD[T: ClassTag](var sqlRDD: RDD[T], @transient conf: SQLConf) extends RDD[T](sqlRDD) {
+ private val sqlConfigs = conf.getAllConfs
+ private lazy val sqlConfExecutorSide = {
+ val newConf = new SQLConf()
+ sqlConfigs.foreach { case (k, v) => newConf.setConfString(k, v) }
+ newConf
+ }
+
+ override val partitioner = firstParent[InternalRow].partitioner
+
+ override def getPartitions: Array[Partition] = firstParent[InternalRow].partitions
+
+ override def compute(split: Partition, context: TaskContext): Iterator[T] = {
+ // If we are in the context of a tracked SQL operation, `SQLExecution.EXECUTION_ID_KEY` is set
+ // and we have nothing to do here. Otherwise, we use the `SQLConf` captured at the creation of
+ // this RDD.
+ if (context.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) == null) {
+ SQLConf.withExistingConf(sqlConfExecutorSide) {
Review Comment:
@alexeykudinkin I was asking the latter.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #6352:
URL: https://github.com/apache/hudi/pull/6352#issuecomment-1223689061
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10715",
"triggerID" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"triggerType" : "PUSH"
}, {
"hash" : "285ed43d8ca7b525c3bd5334a0acd19ec2c7757c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "285ed43d8ca7b525c3bd5334a0acd19ec2c7757c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10715)
* 285ed43d8ca7b525c3bd5334a0acd19ec2c7757c UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #6352:
URL: https://github.com/apache/hudi/pull/6352#issuecomment-1210100544
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #6352:
URL: https://github.com/apache/hudi/pull/6352#issuecomment-1223853439
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10715",
"triggerID" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"triggerType" : "PUSH"
}, {
"hash" : "285ed43d8ca7b525c3bd5334a0acd19ec2c7757c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10895",
"triggerID" : "285ed43d8ca7b525c3bd5334a0acd19ec2c7757c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 285ed43d8ca7b525c3bd5334a0acd19ec2c7757c Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10895)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua merged pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
yihua merged PR #6352:
URL: https://github.com/apache/hudi/pull/6352
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on code in PR #6352:
URL: https://github.com/apache/hudi/pull/6352#discussion_r953174810
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/SQLConfInjectingRDD.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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
+
+import org.apache.spark.{Partition, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.internal.SQLConf
+
+import scala.reflect.ClassTag
+
+/**
+ * NOTE: This is a generalized version of of Spark's [[SQLExecutionRDD]]
+ *
+ * It is just a wrapper over [[sqlRDD]] which sets and makes effective all the configs from the
+ * captured [[SQLConf]]
+ *
+ * @param sqlRDD the `RDD` generated by the SQL plan
+ * @param conf the `SQLConf` to apply to the execution of the SQL plan
+ */
+class SQLConfInjectingRDD[T: ClassTag](var sqlRDD: RDD[T], @transient conf: SQLConf) extends RDD[T](sqlRDD) {
+ private val sqlConfigs = conf.getAllConfs
+ private lazy val sqlConfExecutorSide = {
+ val newConf = new SQLConf()
+ sqlConfigs.foreach { case (k, v) => newConf.setConfString(k, v) }
+ newConf
+ }
+
+ override val partitioner = firstParent[InternalRow].partitioner
+
+ override def getPartitions: Array[Partition] = firstParent[InternalRow].partitions
+
+ override def compute(split: Partition, context: TaskContext): Iterator[T] = {
+ // If we are in the context of a tracked SQL operation, `SQLExecution.EXECUTION_ID_KEY` is set
+ // and we have nothing to do here. Otherwise, we use the `SQLConf` captured at the creation of
+ // this RDD.
+ if (context.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) == null) {
+ SQLConf.withExistingConf(sqlConfExecutorSide) {
Review Comment:
Yes, it will propagate to all RDDs in the execution chain (up to a shuffling point)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #6352:
URL: https://github.com/apache/hudi/pull/6352#issuecomment-1223695207
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10715",
"triggerID" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"triggerType" : "PUSH"
}, {
"hash" : "285ed43d8ca7b525c3bd5334a0acd19ec2c7757c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10895",
"triggerID" : "285ed43d8ca7b525c3bd5334a0acd19ec2c7757c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10715)
* 285ed43d8ca7b525c3bd5334a0acd19ec2c7757c Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10895)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #6352:
URL: https://github.com/apache/hudi/pull/6352#issuecomment-1210186133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10715",
"triggerID" : "e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e86c8aa19d9ee3f7315b00b8b79f50e3e13cb203 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10715)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on code in PR #6352:
URL: https://github.com/apache/hudi/pull/6352#discussion_r952852924
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/SQLConfInjectingRDD.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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
+
+import org.apache.spark.{Partition, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.internal.SQLConf
+
+import scala.reflect.ClassTag
+
+/**
+ * NOTE: This is a generalized version of of Spark's [[SQLExecutionRDD]]
+ *
+ * It is just a wrapper over [[sqlRDD]] which sets and makes effective all the configs from the
+ * captured [[SQLConf]]
+ *
+ * @param sqlRDD the `RDD` generated by the SQL plan
+ * @param conf the `SQLConf` to apply to the execution of the SQL plan
+ */
+class SQLConfInjectingRDD[T: ClassTag](var sqlRDD: RDD[T], @transient conf: SQLConf) extends RDD[T](sqlRDD) {
+ private val sqlConfigs = conf.getAllConfs
+ private lazy val sqlConfExecutorSide = {
+ val newConf = new SQLConf()
+ sqlConfigs.foreach { case (k, v) => newConf.setConfString(k, v) }
+ newConf
+ }
+
+ override val partitioner = firstParent[InternalRow].partitioner
+
+ override def getPartitions: Array[Partition] = firstParent[InternalRow].partitions
+
+ override def compute(split: Partition, context: TaskContext): Iterator[T] = {
+ // If we are in the context of a tracked SQL operation, `SQLExecution.EXECUTION_ID_KEY` is set
+ // and we have nothing to do here. Otherwise, we use the `SQLConf` captured at the creation of
+ // this RDD.
+ if (context.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) == null) {
+ SQLConf.withExistingConf(sqlConfExecutorSide) {
Review Comment:
Not sure i understood your question: do you mean whether we're wrapping any other chained RDD, or whether the SQLConf will get propagated to every other chained RDD?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua commented on a diff in pull request #6352: [HUDI-4584] Fixing `SQLConf` not being propagated to executor
Posted by GitBox <gi...@apache.org>.
yihua commented on code in PR #6352:
URL: https://github.com/apache/hudi/pull/6352#discussion_r952255031
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/SQLConfInjectingRDD.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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
+
+import org.apache.spark.{Partition, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.internal.SQLConf
+
+import scala.reflect.ClassTag
+
+/**
+ * NOTE: This is a generalized version of of Spark's [[SQLExecutionRDD]]
+ *
+ * It is just a wrapper over [[sqlRDD]] which sets and makes effective all the configs from the
+ * captured [[SQLConf]]
+ *
+ * @param sqlRDD the `RDD` generated by the SQL plan
+ * @param conf the `SQLConf` to apply to the execution of the SQL plan
+ */
+class SQLConfInjectingRDD[T: ClassTag](var sqlRDD: RDD[T], @transient conf: SQLConf) extends RDD[T](sqlRDD) {
+ private val sqlConfigs = conf.getAllConfs
+ private lazy val sqlConfExecutorSide = {
+ val newConf = new SQLConf()
+ sqlConfigs.foreach { case (k, v) => newConf.setConfString(k, v) }
+ newConf
+ }
+
+ override val partitioner = firstParent[InternalRow].partitioner
+
+ override def getPartitions: Array[Partition] = firstParent[InternalRow].partitions
+
+ override def compute(split: Partition, context: TaskContext): Iterator[T] = {
+ // If we are in the context of a tracked SQL operation, `SQLExecution.EXECUTION_ID_KEY` is set
+ // and we have nothing to do here. Otherwise, we use the `SQLConf` captured at the creation of
+ // this RDD.
+ if (context.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) == null) {
+ SQLConf.withExistingConf(sqlConfExecutorSide) {
Review Comment:
Is there any performance hit by doing this? Based on the function signature, this is done per Spark partition.
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/SQLConfInjectingRDD.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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
+
+import org.apache.spark.{Partition, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.internal.SQLConf
+
+import scala.reflect.ClassTag
+
+/**
+ * NOTE: This is a generalized version of of Spark's [[SQLExecutionRDD]]
+ *
+ * It is just a wrapper over [[sqlRDD]] which sets and makes effective all the configs from the
+ * captured [[SQLConf]]
+ *
+ * @param sqlRDD the `RDD` generated by the SQL plan
+ * @param conf the `SQLConf` to apply to the execution of the SQL plan
+ */
+class SQLConfInjectingRDD[T: ClassTag](var sqlRDD: RDD[T], @transient conf: SQLConf) extends RDD[T](sqlRDD) {
+ private val sqlConfigs = conf.getAllConfs
+ private lazy val sqlConfExecutorSide = {
+ val newConf = new SQLConf()
+ sqlConfigs.foreach { case (k, v) => newConf.setConfString(k, v) }
+ newConf
+ }
+
+ override val partitioner = firstParent[InternalRow].partitioner
+
+ override def getPartitions: Array[Partition] = firstParent[InternalRow].partitions
+
+ override def compute(split: Partition, context: TaskContext): Iterator[T] = {
+ // If we are in the context of a tracked SQL operation, `SQLExecution.EXECUTION_ID_KEY` is set
+ // and we have nothing to do here. Otherwise, we use the `SQLConf` captured at the creation of
+ // this RDD.
+ if (context.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) == null) {
+ SQLConf.withExistingConf(sqlConfExecutorSide) {
Review Comment:
To clarify, does this happen for all subsequent chained RDD transformations?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org