KnightChess commented on code in PR #11568:
URL: https://github.com/apache/hudi/pull/11568#discussion_r1667357115
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -408,12 +408,20 @@ case class ResolveImplementationsEarly() extends
Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = {
plan match {
// Convert to InsertIntoHoodieTableCommand
- case iis @ MatchInsertIntoStatement(relation @ ResolvesToHudiTable(_),
partition, query, overwrite, _) if query.resolved =>
+ case iis @ MatchInsertIntoStatement(relation @ ResolvesToHudiTable(_),
userSpecifiedCols, partition, query, overwrite, _) if query.resolved =>
relation match {
// NOTE: In Spark >= 3.2, Hudi relations will be resolved as
[[DataSourceV2Relation]]s by default;
// However, currently, fallback will be applied downgrading
them to V1 relations, hence
// we need to check whether we could proceed here, or has to
wait until fallback rule kicks in
- case lr: LogicalRelation => new InsertIntoHoodieTableCommand(lr,
query, partition, overwrite)
+ case lr: LogicalRelation =>
+ // Create a project if this is an INSERT INTO query with specified
cols.
+ val projectByUserSpecified = if (userSpecifiedCols.nonEmpty) {
+ assert(lr.catalogTable.isDefined, "Missing catalog table")
Review Comment:
done
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieCatalystPlansUtils.scala:
##########
@@ -112,7 +113,7 @@ trait HoodieCatalystPlansUtils {
* Decomposes [[InsertIntoStatement]] into its arguments allowing to
accommodate for API
* changes in Spark 3.3
*/
- def unapplyInsertIntoStatement(plan: LogicalPlan): Option[(LogicalPlan,
Map[String, Option[String]], LogicalPlan, Boolean, Boolean)]
+ def unapplyInsertIntoStatement(plan: LogicalPlan): Option[(LogicalPlan,
Seq[String], Map[String, Option[String]], LogicalPlan, Boolean, Boolean)]
Review Comment:
done
##########
hudi-spark-datasource/hudi-spark2/src/main/scala/org/apache/spark/sql/HoodieSpark2CatalystPlanUtils.scala:
##########
@@ -61,10 +61,10 @@ object HoodieSpark2CatalystPlanUtils extends
HoodieCatalystPlansUtils {
Join(left, right, joinType, None)
}
- override def unapplyInsertIntoStatement(plan: LogicalPlan):
Option[(LogicalPlan, Map[String, Option[String]], LogicalPlan, Boolean,
Boolean)] = {
+ override def unapplyInsertIntoStatement(plan: LogicalPlan):
Option[(LogicalPlan, Seq[String], Map[String, Option[String]], LogicalPlan,
Boolean, Boolean)] = {
plan match {
case InsertIntoTable(table, partition, query, overwrite,
ifPartitionNotExists) =>
- Some((table, partition, query, overwrite, ifPartitionNotExists))
+ Some((table, Seq.empty, partition, query, overwrite,
ifPartitionNotExists))
Review Comment:
I think it unnecessary, because it is not supported at the sql grammar level
##########
hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/HoodieSpark3CatalystPlanUtils.scala:
##########
@@ -56,15 +57,6 @@ trait HoodieSpark3CatalystPlanUtils extends
HoodieCatalystPlansUtils {
Join(left, right, joinType, None, JoinHint.NONE)
}
- override def unapplyInsertIntoStatement(plan: LogicalPlan):
Option[(LogicalPlan, Map[String, Option[String]], LogicalPlan, Boolean,
Boolean)] = {
- plan match {
- case insert: InsertIntoStatement =>
Review Comment:
every subclass has it own impl, so I remove it
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
