Open zohar-plutoflume opened 1 year ago
zohar-plutoflume
commented
1 year ago after looking into it a bit more I think iceberg is not using the spark dynamic filtering but does it on its own, using the RowLevelCommandDynamicPruning however it seems like it has a similar condition check: private def isCandidate(command: RowLevelCommand): Boolean = command.condition match { case Some(cond) if cond != Literal.TrueLiteral => true case _ => false }
and I wonder what would be a work around it
Grumi
commented
8 months ago Hello Zohar,
I am facing the same problem after I upgraded from spark 3.3.0 ->3.3.2 and iceber 1.1.0 -> 1.3.0. Have you been able to solve the problem?
Thank you, Marius
github-actions[bot]
commented
2 days ago This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.
Query engine
spark 3.3.1 iceberg 1.1 (emr 6.10)
Question
Hi, wanted to point to something that was introduced in spark 3.3, https://issues.apache.org/jira/browse/SPARK-38148 , the issue with this is that when we are using a merge command and in case we have a static partition as part of the command, this translates to a join query which does not utilise dynamic filtering and caused our jobs to run much slower. for example this merge into command: f""" MERGE INTO {output_table.catalog_table_ref} {TARGET} USING {TMP_VIEW} {SOURCE} ON {join_str} AND {TARGET}.triggered is true WHEN MATCHED THEN UPDATE SET {update_col_string} """ if we were to use spark joins directly we could just filter on the target table , but as we use the merge api we need to provide the iceberg target so we have to add the static partition condition as part of the on.
one option is putting it in WHEN MATCHED AND {TARGET}.triggered is true , but I think it will not push down the triggered = true to the target table.
any suggestion of how to make the merge use the dynamic filtering like before the spark 3.3.1 upgrade would be very appreciated.
I'm thinking if its a spark issue, where they need to introduce a conf which will decide if to drop the dynamic partitioning or not, or if its an iceberg issue cause its the merge into api.
after the change we can see the merge creates a join as follows:
before the change, the join was in two steps where the first step it tried to find the ids needed for the join only and after that another join based on those ids. the difference in performance were due to the fact that before the upgrade , the first join projected only the needed columns for finding out the id which resulted in much smaller shuffle data overall.
and here is the physical plan of after the upgrade:
and before the upgrade: