You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "Zouxxyy (via GitHub)" <gi...@apache.org> on 2023/08/07 15:01:32 UTC
[GitHub] [incubator-paimon] Zouxxyy opened a new pull request, #1756: [spark] supports dynamic insert overwrite
Zouxxyy opened a new pull request, #1756:
URL: https://github.com/apache/incubator-paimon/pull/1756
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] YannByron commented on a diff in pull request #1756: [spark] supports dynamic insert overwrite
Posted by "YannByron (via GitHub)" <gi...@apache.org>.
YannByron commented on code in PR #1756:
URL: https://github.com/apache/incubator-paimon/pull/1756#discussion_r1286896243
##########
paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTest.scala:
##########
@@ -215,4 +214,45 @@ class InsertOverwriteTest extends PaimonSparkTestBase {
Row(4, "d", sv2.value) :: Row(5, "e", sv1.value) :: Row(7, "g", sv1.value) :: Nil)
}
}
+
+ writeModes.foreach {
+ writeMode =>
+ bucketModes.foreach {
+ bucket =>
+ test(
+ s"dynamic insert overwrite single-partitioned table: write-mode: $writeMode, bucket: $bucket") {
+ val primaryKeysProp = if (writeMode == CHANGE_LOG) {
+ "'primary-key'='a,b',"
+ } else {
+ ""
+ }
+
+ spark.sql(
+ s"""
+ |CREATE TABLE T (a INT, b INT, c STRING)
+ |TBLPROPERTIES ($primaryKeysProp 'write-mode'='${writeMode.toString}', 'bucket'='$bucket')
+ |PARTITIONED BY (a)
+ |""".stripMargin)
+
+ spark.sql("INSERT INTO T values (1, 1, '1'), (2, 2, '2')")
+ checkAnswer(
+ spark.sql("SELECT * FROM T ORDER BY a, b"),
+ Row(1, 1, "1") :: Row(2, 2, "2") :: Nil)
+
+ // overwrite the whole table
+ spark.sql("INSERT OVERWRITE T VALUES (1, 3, '3'), (2, 4, '4')")
+ checkAnswer(
+ spark.sql("SELECT * FROM T ORDER BY a, b"),
+ Row(1, 3, "3") :: Row(2, 4, "4") :: Nil)
+
+ withSQLConf("spark.sql.sources.partitionOverwriteMode" -> "dynamic") {
+ // overwrite the a=1 partition
+ spark.sql("INSERT OVERWRITE T VALUES (1, 5, '5'), (1, 7, '7')")
+ checkAnswer(
+ spark.sql("SELECT * FROM T ORDER BY a, b"),
+ Row(1, 5, "5") :: Row(1, 7, "7") :: Row(2, 4, "4") :: Nil)
+ }
+ }
+ }
+ }
Review Comment:
please enrich multi-partitions case.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] YannByron commented on pull request #1756: [spark] supports dynamic insert overwrite
Posted by "YannByron (via GitHub)" <gi...@apache.org>.
YannByron commented on PR #1756:
URL: https://github.com/apache/incubator-paimon/pull/1756#issuecomment-1670561965
Looks good. @JingsongLi can you take a look?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Zouxxyy commented on a diff in pull request #1756: [spark] supports dynamic insert overwrite
Posted by "Zouxxyy (via GitHub)" <gi...@apache.org>.
Zouxxyy commented on code in PR #1756:
URL: https://github.com/apache/incubator-paimon/pull/1756#discussion_r1287862568
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonAnalysis.scala:
##########
@@ -28,6 +32,21 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
case func: PaimonTableValueFunction if func.args.forall(_.resolved) =>
PaimonTableValuedFunctions.resolvePaimonTableValuedFunction(session, func)
+ case o @ OverwritePartitionsDynamicPaimon(r, d) if o.resolved =>
+ PaimonDynamicPartitionOverwriteCommand(r, d, o.query, o.writeOptions, o.isByName)
}
+}
+object OverwritePartitionsDynamicPaimon {
+ def unapply(o: OverwritePartitionsDynamic): Option[(DataSourceV2Relation, FileStoreTable)] = {
+ if (o.query.resolved) {
+ o.table match {
+ case r: DataSourceV2Relation if r.table.isInstanceOf[SparkTable] =>
Review Comment:
@YannByron The constructor of `DataSourceV2Relation` may change, the current code is more conducive to multi-version compatibility
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] YannByron commented on a diff in pull request #1756: [spark] supports dynamic insert overwrite
Posted by "YannByron (via GitHub)" <gi...@apache.org>.
YannByron commented on code in PR #1756:
URL: https://github.com/apache/incubator-paimon/pull/1756#discussion_r1286889937
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonAnalysis.scala:
##########
@@ -18,8 +18,10 @@
package org.apache.paimon.spark
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OverwritePartitionsDynamic}
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.paimon.commands.PaimonDynamicPartitionOverwriteCommand
Review Comment:
Put paimon's import clause in the front of others.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Zouxxyy commented on a diff in pull request #1756: [spark] supports dynamic insert overwrite
Posted by "Zouxxyy (via GitHub)" <gi...@apache.org>.
Zouxxyy commented on code in PR #1756:
URL: https://github.com/apache/incubator-paimon/pull/1756#discussion_r1287864237
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonAnalysis.scala:
##########
@@ -18,8 +18,10 @@
package org.apache.paimon.spark
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OverwritePartitionsDynamic}
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.paimon.commands.PaimonDynamicPartitionOverwriteCommand
Review Comment:
We have to put the class under `org.apache.spark.sql` so that some private methods can be used, and the current import clause is the result of automatic formatting
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on pull request #1756: [spark] supports dynamic insert overwrite
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #1756:
URL: https://github.com/apache/incubator-paimon/pull/1756#issuecomment-1670570347
> Looks good. @JingsongLi can you take a look and trigger CI?
Testing has passed in @Zouxxyy 's repo: https://github.com/Zouxxyy/incubator-paimon/tree/dev/support-dynamic
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi merged pull request #1756: [spark] supports dynamic insert overwrite
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #1756:
URL: https://github.com/apache/incubator-paimon/pull/1756
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] YannByron commented on a diff in pull request #1756: [spark] supports dynamic insert overwrite
Posted by "YannByron (via GitHub)" <gi...@apache.org>.
YannByron commented on code in PR #1756:
URL: https://github.com/apache/incubator-paimon/pull/1756#discussion_r1287854138
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonAnalysis.scala:
##########
@@ -28,6 +32,21 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
case func: PaimonTableValueFunction if func.args.forall(_.resolved) =>
PaimonTableValuedFunctions.resolvePaimonTableValuedFunction(session, func)
+ case o @ OverwritePartitionsDynamicPaimon(r, d) if o.resolved =>
+ PaimonDynamicPartitionOverwriteCommand(r, d, o.query, o.writeOptions, o.isByName)
}
+}
+object OverwritePartitionsDynamicPaimon {
+ def unapply(o: OverwritePartitionsDynamic): Option[(DataSourceV2Relation, FileStoreTable)] = {
+ if (o.query.resolved) {
+ o.table match {
+ case r: DataSourceV2Relation if r.table.isInstanceOf[SparkTable] =>
Review Comment:
case r @ DataSourceV2Relation(r: SparkTable, _, _, ...) =>
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] YannByron commented on a diff in pull request #1756: [spark] supports dynamic insert overwrite
Posted by "YannByron (via GitHub)" <gi...@apache.org>.
YannByron commented on code in PR #1756:
URL: https://github.com/apache/incubator-paimon/pull/1756#discussion_r1286891010
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonAnalysis.scala:
##########
@@ -28,6 +30,21 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
case func: PaimonTableValueFunction if func.args.forall(_.resolved) =>
PaimonTableValuedFunctions.resolvePaimonTableValuedFunction(session, func)
+ case o @ OverwritePartitionsDynamicPaimon(r, d) if o.resolved =>
+ PaimonDynamicPartitionOverwriteCommand(r, d, o.query, o.writeOptions, o.isByName)
}
+}
+object OverwritePartitionsDynamicPaimon {
+ def unapply(o: OverwritePartitionsDynamic): Option[(DataSourceV2Relation, SparkTable)] = {
Review Comment:
Suggest that return Paimon `FileStoreTable` instead of `SparkTable`.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org