CometShuffleManager treats Spark's own shuffles differently from SortShuffleManager

Đang mở
  • Có gì không ổn?
#7,488 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 2 ngày

Chưa có ai nhận issue này.

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
55/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
scala, spark
Lĩnh vực
backend, performance

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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.

Ngôn ngữ chính
Scala
Star
1.3k
Fork
414
Merge trung bình
3 ngày 5 giờ
Pull request đã merge (30 ngày)
210

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/datafusion-comet

Tất cả issue của apache/datafusion-comet

Issue tương tự

Thêm issue về Scala

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.