[sushanb]

Summary

Fixes four error-handling and resumption boundary issues in DefaultRowReader, AsyncRowReader, and DataConnectionImpl::ReadRow / AsyncReadRow:

1. Negative rows_limit silently returns 0 rows (OK): Because rows_limit_ != NO_ROWS_LIMIT && rows_limit_ <= rows_count_ (NO_ROWS_LIMIT == 0) evaluated true for any negative rows_limit when rows_count_ == 0, negative rows_limit values suppressed errors and terminated as an empty OK scan instead of failing fast with client-side INVALID_ARGUMENT.
2. Swallowed cancellation and parser errors at scan boundaries: When rows_limit was reached or row_set_ became empty after the last delivered/scanned row, DefaultRowReader::Advance and AsyncRowReader::OnStreamFinished unconditionally overwrote status with OK, masking user cancellations (Cancel / OnRow returning false or throwing), mid-stream ReadRowsParser protocol errors (INTERNAL), and HandleEndOfStream errors on a clean EOF when rows_limit had not been reached.
3. Premature last_scanned_row_key overwrite in DefaultRowReader::NextChunk: Updating last_read_row_key_ from response_.last_scanned_row_key() immediately on stream_->Read(&response_) (before consuming response_.chunks()) allowed Advance() to overwrite last_read_row_key_ with an earlier chunk's parsed_row.row_key(), causing retries to re-request already-scanned rows up to last_scanned_row_key.
4. Spurious client cancellation and post-row error reporting in ReadRow / AsyncReadRow: Returning make_ready_future(false) from AsyncReadRowHandler::OnRow triggered AsyncRowReader::Cancel("User cancelled") on every single-row read (rows_limit = 1); returning true lets the rows_limit = 1 stream close cleanly with OK, while both ReadRow and AsyncReadRow propagate post-row parser errors (INTERNAL) and still reject unexpected extra rows.

Changes

• Validate rows_limit_ < 0 in DefaultRowReader::Advance and AsyncRowReader::MakeRequest, returning INVALID_ARGUMENT without issuing an RPC.
• Track parser_failed_ / user_cancelled_ (including HandleEndOfStream failures on clean EOF when !ReachedRowsLimit(...)) and preserve those statuses at rows_limit and empty-RowSet scan boundaries while still ignoring benign trailing transport errors once all requested rows have been delivered.
• Apply response_.last_scanned_row_key() in DefaultRowReader::NextChunk after all chunks in response_ have been consumed.
• Propagate post-row non-OK statuses in DataConnectionImpl::ReadRow (if (!*it) return std::move(*it).status();) and DataConnectionImpl::AsyncReadRow, while draining rows_limit = 1 streams cleanly without cancelling.
• Add unit tests across default_row_reader_test.cc, async_row_reader_test.cc, and data_connection_impl_test.cc.

[gemini-code-assist]

Code Review

This pull request refactors the row reading logic in both AsyncRowReader and DefaultRowReader to correctly handle edge cases around row limits, row set boundaries, and parser failures. Specifically, it ensures that parser failures are not swallowed when a row limit is reached or a row set is exhausted, and it adds validation to reject negative row limits with an invalid argument error. It also updates DataConnectionImpl to properly drain streams when a single row is requested and adds comprehensive unit tests covering these scenarios. There are no review comments provided, so I have no additional feedback to offer.

[codecov]

Codecov Report

❌ Patch coverage is 98.27128% with 13 lines in your changes missing coverage. Please review.
✅ Project coverage is 92.50%. Comparing base (5dcdef5) to head (da1ea01).
⚠️ Report is 20 commits behind head on main.

Files with missing lines	Patch %	Lines
...cloud/bigtable/internal/default_row_reader_test.cc	97.12%	6 Missing ⚠️
...oud/bigtable/internal/data_connection_impl_test.cc	97.81%	3 Missing ⚠️
...ogle/cloud/bigtable/internal/default_row_reader.cc	91.89%	3 Missing ⚠️
google/cloud/bigtable/internal/async_row_reader.cc	96.15%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #16660      +/-   ##
==========================================
- Coverage   92.53%   92.50%   -0.03%     
==========================================
  Files        2269     2269              
  Lines      222896   223604     +708     
==========================================
+ Hits       206250   206853     +603     
- Misses      16646    16751     +105     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[scotthart]

1. [Bug / Asymmetry] Synchronous DataConnectionImpl::ReadRow misreports post-row parser errors as "returned more than one row"

In data_connection_impl.cc, ReadRow drains the rows_limit = 1 reader after reading the first row by checking if (++it != reader.end()):

  auto it = reader.begin();
  if (it == reader.end()) return std::make_pair(false, bigtable::Row("", {}));
  if (!*it) return it->status();
  auto result = std::make_pair(true, std::move(**it));
  if (++it != reader.end()) {
    return internal::InternalError(
        "internal error - RowReader returned more than one row in ReadRow(, "
        ")",
        GCP_ERROR_INFO());
  }
  return result;

Previously, DefaultRowReader::Advance() swallowed all errors once rows_count_ == 1 (rows_limit = 1), so ++it != reader.end() could only happen if a second Row was yielded. With this PR, if a trailing chunk in the rows_limit = 1 stream fails in the parser (parser_failed_ == true), ++it is != reader.end() and *it holds the non-OK parser Status. ReadRow will discard it->status() and return "internal error - RowReader returned more than one row in ReadRow(, )" instead of the actual parser error (unlike AsyncReadRow, which now propagates status via if (!status.ok())).

Suggestion: Check if (!*it) return std::move(*it).status(); inside if (++it != reader.end()):

  if (++it != reader.end()) {
    if (!*it) return std::move(*it).status();
    return internal::InternalError(
        "internal error - RowReader returned more than one row in ReadRow(, "
        ")",
        GCP_ERROR_INFO());
  }

---

2. [Edge Case] HandleEndOfStream parser errors at a clean EOF are still swallowed when row_set_.IsEmpty() and !reached_rows_limit

In both async_row_reader.cc (OnStreamFinished) and default_row_reader.cc (AdvanceOrFail), parser_->HandleEndOfStream(...) is only used when the underlying gRPC stream itself finished with Status::OK, and it only fails when the stream ended with an uncommitted cell or row ("end of stream with unfinished cell" / "end of stream with unfinished row").

Because parser_failed_ is only set in HandleChunk / Next and not when HandleEndOfStream fails:

• If reached_rows_limit is true, ignoring the HandleEndOfStream error is intentional (tested by ParserEofDoesntFailOnUnfinishedRowIfRowLimit and UnfinishedRowAtEofIgnoredIfRowLimitReached), since rows_limit_ committed rows were already received and any partial (rows_limit_ + 1)-th row at EOF is unneeded.
• However, if reached_rows_limit is false and row_set_.IsEmpty() becomes true (for example, a request for RowSet("r1") or RowSet("r1", "r2") receives an uncommitted chunk for "r2" with commit_row = false and last_scanned_row_key = "r2", followed by a clean stream EOF), row_set_.IsEmpty() is true while parser_failed_ is false. Both readers will overwrite the HandleEndOfStream INTERNAL error with Status{} (OK), silently dropping the truncated row.

Suggestion: Only ignore HandleEndOfStream failures when reached_rows_limit is true (or set parser_failed_ = true when HandleEndOfStream fails on a clean EOF and !reached_rows_limit).

---

3. [Nit / Dead Condition] operation_cancelled_ check inside DefaultRowReader::Advance() loop is unreachable

In DefaultRowReader::Advance() (google/cloud/bigtable/internal/default_row_reader.cc lines 125–155):

  if (operation_cancelled_) {
    return internal::CancelledError(
        "call cancelled",
        GCP_ERROR_INFO().WithMetadata("gl-cpp.error.origin", "client"));
  }
  ...
  while (true) {
    ...
    if (ReachedRowsLimit(rows_limit_, rows_count_) || row_set_.IsEmpty()) {
      if (operation_cancelled_ || parser_failed_) return status;
      last_status_ = Status{};
      return Status{};
    }

Unlike AsyncRowReader (where user_cancelled_ is set asynchronously when on_row_ returns false mid-stream), DefaultRowReader is synchronous: operation_cancelled_ is checked at the top of Advance() before the loop and is never modified by AdvanceOrFail(), NextChunk(), or MakeRequest(). Checking operation_cancelled_ again inside the loop is redundant and can be simplified to if (parser_failed_) return status;.

---

4. [Testing / Codecov] Missing unit tests for new branches in DataConnectionImpl::AsyncReadRow

Codecov flags google/cloud/bigtable/internal/data_connection_impl.cc at 69.23% patch coverage because the new branches in AsyncReadRowHandler (extra_row_received_ when >1 row is delivered, and !status.ok() when a parser error occurs after receiving a row) are not exercised in data_connection_impl_test.cc.

(Note: The remaining missing patch coverage lines reported by Codecov in default_row_reader.cc and async_row_reader.cc are on the if (!grpc_status.ok()) / if (!parser_status.ok()) checks immediately after parser_->Next(...) / DrainParser(), which cannot fail when guarded by while (parser_->HasNext()).)

Suggestion: Add unit tests in data_connection_impl_test.cc for:

1. AsyncReadRow (and ReadRow) when the stream yields a valid row followed by a malformed chunk (verifying StatusCode::kInternal is returned rather than OK).
2. AsyncReadRow when the mock stream returns two committed rows despite rows_limit = 1 (verifying extra_row_received_ returns StatusCode::kInternal with "AsyncRowReader returned more than one row").

---

[scotthart]

See PR level comment.

[sushanb]

Thanks for the catch! Addressed all 4 items in 662a5d6b2d:

1. Propagated post-row parser errors in DataConnectionImpl::ReadRow: Added if (!*it) return std::move(*it).status(); inside if (++it != reader.end()) so trailing parser errors on a rows_limit = 1 stream return the actual parser status instead of "returned more than one row" (ReadRowParserFailureAfterRow).
2. Preserved HandleEndOfStream errors on clean EOF when !reached_rows_limit: Set parser_failed_ = true in both DefaultRowReader::AdvanceOrFail and AsyncRowReader::OnStreamFinished when the stream ends cleanly (OK) with a HandleEndOfStream error (unfinished cell/row) and rows_limit has not been reached, so an exhausted row_set_ no longer overwrites the truncated-row INTERNAL error with OK. Added UnfinishedRowAtCleanEofNotSwallowedWhenRowSetExhausted in both default_row_reader_test.cc and async_row_reader_test.cc.
3. Removed redundant operation_cancelled_ check inside DefaultRowReader::Advance loop: Simplified if (operation_cancelled_ || parser_failed_) return status; to if (parser_failed_) return status;.
4. Added DataConnectionImpl unit tests (data_connection_impl_test.cc): Added ReadRowParserFailureAfterRow, AsyncReadRowParserFailureAfterRow, and AsyncReadRowExtraRowFails to cover the post-row error and extra_row_received_ branches.

/gemini review

[gemini-code-assist]

Code Review

This pull request refactors AsyncRowReader, DefaultRowReader, and DataConnectionImpl to properly handle negative row limits, track parser failures, and ensure that parser errors are not swallowed when row limits are reached or row sets are exhausted. It also adds comprehensive unit tests for these edge cases. Feedback on the changes suggests a minor improvement in data_connection_impl.cc to use it->status() instead of std::move(*it).status() for consistency and to avoid an unnecessary move on a dereferenced iterator.