Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: Avoid exception caused by broadcasting empty result #92

Merged
merged 2 commits into from
Feb 23, 2024
Merged

fix: Avoid exception caused by broadcasting empty result #92

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8dd2849
fix: Avoid exception caused by broadcasting empty result
wForget Feb 22, 2024
5ecac7e
add test
wForget Feb 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ class CometShuffleManager(conf: SparkConf) extends ShuffleManager with Logging {
" Shuffle will continue to spill to disk when necessary.")
}

private val sortShuffleManager = new SortShuffleManager(conf);
private val sortShuffleManager = new SortShuffleManager(conf)

/**
* A mapping from shuffle ids to the task ids of mappers producing output for those shuffles.
Expand Down
6 changes: 5 additions & 1 deletion spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,11 @@ abstract class CometExec extends CometPlan {

out.flush()
out.close()
Iterator((count, cbbos.toChunkedByteBuffer))
if (out.size() > 0) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is due to the extra bytes being written in the encode compressed stream, causing it to be determined not empty in CometExec$.decodeBatches.

If there are extra bytes written, is out.size() 0?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, I see. The compressed stream is wrapped in out. When it is flushed, it might write compression related bytes which are not counted in out.size().

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about we check count > 0 instead?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that is same.

Iterator((count, cbbos.toChunkedByteBuffer))
} else {
Iterator((count, new ChunkedByteBuffer(Array.empty[ByteBuffer])))
}
}
}

Expand Down
19 changes: 19 additions & 0 deletions spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,25 @@ class CometExecSuite extends CometTestBase {
}
}

test("CometBroadcastExchangeExec: empty broadcast") {
withSQLConf(CometConf.COMET_EXEC_BROADCAST_ENABLED.key -> "true") {
withParquetTable((0 until 5).map(i => (i, i + 1)), "tbl_a") {
withParquetTable((0 until 5).map(i => (i, i + 1)), "tbl_b") {
val df = sql(
"SELECT /*+ BROADCAST(a) */ *" +
" FROM (SELECT * FROM tbl_a WHERE _1 < 0) a JOIN tbl_b b" +
" ON a._1 = b._1")
val nativeBroadcast = find(df.queryExecution.executedPlan) {
case _: CometBroadcastExchangeExec => true
case _ => false
}.get.asInstanceOf[CometBroadcastExchangeExec]
val rows = nativeBroadcast.executeCollect()
assert(rows.isEmpty)
}
}
}
}

test("CometExec.executeColumnarCollectIterator can collect ColumnarBatch results") {
withSQLConf(
CometConf.COMET_EXEC_ENABLED.key -> "true",
Expand Down