SubhamSinghal commented on code in PR #25467:
URL: https://github.com/apache/datafusion/pull/25467#discussion_r4052515209
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs:
##########
@@ -409,22 +412,33 @@ impl ExistencePWMJStream {
.load(AtomicOrdering::SeqCst)
.min(buffered_len);
- let sliced = match self.join_type {
+ let (num_rows, columns) = match self.join_type {
JoinType::LeftSemi => {
- buffered_batch.slice(min_marked, buffered_len - min_marked)
+ let sliced =
+ buffered_batch.slice(min_marked, buffered_len -
min_marked);
+ (sliced.num_rows(), sliced.columns().to_vec())
+ }
+ // `LeftMark` keeps every buffered row -- nothing to slice --
and appends
+ // the watermark as a `mark` column instead of using it to
drop rows.
+ JoinType::LeftMark => {
+ let mut columns = buffered_batch.columns().to_vec();
+ columns.push(mark_column(buffered_len, min_marked));
+ (buffered_len, columns)
+ }
+ // `LeftAnti`: the unmarked prefix, which includes every
null-keyed row --
+ // nulls sort first and the watermark never drops below the
buffered null
+ // count.
+ _ => {
Review Comment:
Addressed in 8885ca1914c081f0f5c97984d7248fcf5c18cd7d
##########
datafusion/proto/tests/cases/plans/joins.rs:
##########
@@ -1079,3 +1062,89 @@ async fn roundtrip_planned_piecewise_merge_join() ->
Result<()> {
}
Ok(())
}
+
+/// `roundtrip_test`/`roundtrip_test_and_return` only compare the `Debug`
string of the
+/// before/after plans -- which, per their own doc comment, "often isn't
sufficient to
+/// guarantee that no information is lost during serde because the string
representation of
+/// a plan often only shows a subset of state". `LeftMark`/`RightMark` add no
new field to
+/// encode (`join_type` already selects them from the shared proto enum, see
+/// `join_type_to_proto`/`join_type_from_proto`), so the real risk is not a
missing wire field
+/// but a decoded plan that behaves differently at execution time. This
actually executes both
+/// the original and the roundtripped plan over real data and compares their
output batches
+/// row for row, including the `mark` column.
+#[tokio::test]
+async fn roundtrip_piecewise_merge_join_mark_executes_correctly() ->
Result<()> {
Review Comment:
Addressed in 8885ca1914c081f0f5c97984d7248fcf5c18cd7d
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
