CometShuffleManager treats Spark's own shuffles differently from SortShuffleManager

Abierto
  • ¿Algo no está bien?
#7,488 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 2 días

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
55/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
scala, spark
Área
backend, performance

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

bug requires-triage

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.

Lenguaje dominante
Scala
Estrellas
1.3k
Forks
414
Merge medio
3 d 5 h
PR fusionados (30 d)
210

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/datafusion-comet

Todos los issues de apache/datafusion-comet

Issues similares

Más issues de Scala

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.