Comet Native Executors: run LIMIT and top-K queries with a per-partition limit before the exchange

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

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

@andygrove đang làm issue này rồi.

Từ ngày 1/11/2026.

Đá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
48/100
Loại issue
Tính năng
Độ rõ ràng
Khá rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
scala
Lĩnh vực
distributed-systems

Hướng nghiên cứu

Start with takeOrderedStage in CometNativeExecutorsExec.scala and trace how it handles CometTakeOrderedAndProjectExec; compare that special case with the proposed per-partition limit/top-K, single-partition exchange, and final native stage. Done means both operators are handled by the generic stage-graph builder and only per-partition limited rows cross the exchange.

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

Mô tả

enhancement
What is the problem the feature request solves?

On a Comet Native Executor cluster (#7289), a query with LIMIT, including Dataset.show(), fails: CometCollectLimitExec performs its own shuffle to one partition inside doExecuteColumnar, so it is not a native block, and the driver's stage-graph builder refuses it with "it does not run natively".

CometTakeOrderedAndProjectExec has the same shape. The prototype special-cases it as a stage that reads its child through a single-partition exchange and runs Comet's final top-K plan. Unless the child is a CometLocalTopKExec, the child's whole output crosses that one-partition exchange, where Comet on Spark runs a per-partition top-K first.

Describe the potential solution

Plan both operators as explicit native pieces, as Spark plans LocalLimit, then an exchange, then GlobalLimit: a block with a per-partition limit or top-K, a CometShuffleExchangeExec to a single partition, and a block with the final limit or top-K. The generic stage-graph builder then handles them without a special case, and only the first rows of each partition cross the exchange.

Additional context

The special case is takeOrderedStage in CometNativeExecutorsExec.scala. It calls finalNativePlan(inputPartitions = 2) only to select the sorting branch.

Part of #7288.

Willingness to contribute

Yes. I can contribute this feature independently.

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

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.