asof_join keep input batch max size

Rafał Hibner · Rafał Hibner · commit 19b216d977f8 · 2025-06-04T13:04:47.000Z
diff --git a/cpp/src/arrow/acero/asof_join_node.cc b/cpp/src/arrow/acero/asof_join_node.cc
@@ -1001,6 +1001,7 @@ class AsofJoinNode : public ExecNode {
         dst.Emplace(state_, tolerance_);
         ARROW_ASSIGN_OR_RAISE(bool advanced, lhs.Advance());
         if (!advanced) break;  // if we can't advance LHS, we're done for this batch
+        if (lhs.GetLatestRow() == 0) break;  // Keep lhs batch boundary
       } else {
         if (!rhs_update_state.any_advanced) break;  // need to wait for new data
       }

Original file line number	Diff line number	Diff line change
`@@ -1001,6 +1001,7 @@ class AsofJoinNode : public ExecNode {`
`1001`	`1001`	`dst.Emplace(state_, tolerance_);`
`1002`	`1002`	`ARROW_ASSIGN_OR_RAISE(bool advanced, lhs.Advance());`
`1003`	`1003`	`if (!advanced) break; // if we can't advance LHS, we're done for this batch`
	`1004`	`+ if (lhs.GetLatestRow() == 0) break; // Keep lhs batch boundary`
`1004`	`1005`	`} else {`
`1005`	`1006`	`if (!rhs_update_state.any_advanced) break; // need to wait for new data`
`1006`	`1007`	`}`