WIP: Low shuffle merge implementation. #10753
Conversation
liurenjie1024
force-pushed
the
renjie/lowshufflemerge
branch
from
7e1c587
to
982a926
Compare
There was a problem hiding this comment.
2024 copyrights. Comment applies to other files as well.
| @JsonDeserialize(contentAs = classOf[java.lang.Long]) | ||
| rows: Option[Long] = None, | ||
| @JsonDeserialize(contentAs = classOf[java.lang.Long]) | ||
| files: Option[Long] = None, | ||
| @JsonDeserialize(contentAs = classOf[java.lang.Long]) | ||
| bytes: Option[Long] = None, | ||
| @JsonDeserialize(contentAs = classOf[java.lang.Long]) | ||
| partitions: Option[Long] = None) |
There was a problem hiding this comment.
Why was this reformatted? It doesn't match the project coding style. Comment applies to elsewhere in this file. A reformat of this file makes it much harder to see the actual changes. The reformat is not related to this PR, please revert it.
| val newWrittenFiles = withStatusCode("DELTA", "Writing merged data") { | ||
| writeAllChanges(spark, deltaTxn, filesToRewrite) | ||
| } | ||
| val newWrittenFiles = lowShuffleMerge(spark, deltaTxn, filesToRewrite) |
There was a problem hiding this comment.
There should be a config to control whether low shuffle merge is performed or not, especially on Delta Lake versions that do not have a low shuffle merge implementation (like OSS Delta Lake).
| private def lowShuffleMerge(spark: SparkSession, | ||
| deltaTxn: OptimisticTransaction, | ||
| filesToRewrite: Seq[AddFile]): Seq[FileAction] = { | ||
| val executor = new LowShuffleMergeExecutor(spark, deltaTxn, filesToRewrite, this) |
There was a problem hiding this comment.
We may want to consider using a GpuLowShuffleMergeIntoCommand separate from GpuMergeIntoCommand (both leveraging common code as appropriate) so this can appear in the SQL UI and make it clear when we're using low shuffle vs. not.
|
Closed by #10786 |
No description provided.