CometShuffleManager treats Spark's own shuffles differently from SortShuffleManager

Offen
  • Stimmt etwas nicht?
#7,488 0 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen

Maintainer antworten meist innerhalb von 2 Tagen

Dieses Issue hat noch niemand übernommen.

Bewertung

Schwierigkeit
4/5
Geschätzter Aufwand
3-5 Tage
Anfängerfreundlichkeit
55/100
Issue-Typ
Bug
Klarheit
Klar beschrieben
Aktivitätsstatus
Aktiv
Tech-Stack
scala, spark
Bereich
backend, performance

Rechercherichtung

Read spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleManager.scala and trace how getWriter, unregisterShuffle, and the IndexShuffleBlockResolver share shuffle state. Check the Spark exchange copy behavior described in the issue and compare cleanup with SortShuffleManager. Done when all three reported differences are addressed; the issue gives no named tests or reproduction.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Beschreibung

area:shuffle bug priority:medium

Describe the bug

CometShuffleManager handles shuffles that Comet does not replace differently from Spark's SortShuffleManager. These shuffles are RDD shuffles and Spark exchanges Comet leaves in place. They are not new, but they reach more applications once the plugin sets CometShuffleManager by default on Spark 4.0 and later (#7336). Found in review of that PR.

  1. Extra row copy in Spark exchanges. ShuffleExchangeExec.needToCopyObjectsBeforeShuffle returns true for any manager that is not a SortShuffleManager, so Spark's own exchanges copy every row before the shuffle.
  2. sortShuffleManager.getWriter entries are never released. For a Spark shuffle dependency, getWriter delegates to the inner sortShuffleManager, which records each map task in its own taskIdMapsForShuffle. unregisterShuffle clears only CometShuffleManager's map and never calls sortShuffleManager.unregisterShuffle, so those entries stay on the executor heap until the application ends.
  3. Blocks migrated in during decommissioning are never deleted. The IndexShuffleBlockResolver is built with a fresh ConcurrentHashMap instead of taskIdMapsForShuffle. The resolver records blocks migrated in during decommissioning in that map, so unregisterShuffle never deletes them.

All three are in spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleManager.scala.

Steps to reproduce

Read the code paths above. No reproduction has been run for any of them.

Expected behavior

A shuffle that Comet does not replace behaves as it does under SortShuffleManager: no extra row copy, and no state left behind after unregisterShuffle.

Vorherrschende Sprache
Scala
Sterne
1.3k
Forks
414
Ø Merge
3 T. 4 Std.
Gemergte PRs (30 T.)
202

Entwicklungsumgebung

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus apache/datafusion-comet

Alle Issues in apache/datafusion-comet

Ähnliche Issues

Weitere Issues zu Scala

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.