fix(bigtable): count mutations wrapped from protos towards the row limits

[laughingman7743]

Fixes #14020.

The defect

Mutation.MAX_MUTATIONS and Mutation.MAX_BYTE_SIZE are enforced in addMutation, against
counters only addMutation maintains. Three factories — fromProtoUnsafe(List),
fromProtoUnsafe(Iterable) and fromProto(List) — add to the mutation list directly and leave
both counters at zero, so mutations wrapped from existing protos count towards neither limit.

Measured on 2.80.0, five mutations wrapped from protos and then MAX_MUTATIONS more added:

MAX_MUTATIONS=100000
seeded=5 addedViaSetCell=100000 threw=null
actual mutations in list = 100005

The two limits then diverge, and only one of them is really lost:

• Mutation count is backstopped by RowMutationEntry.toProto() and BulkMutation.add, which
  re-check getMutations().size(). It surfaces — but late, from a different place, as an
  IllegalArgumentException at send time rather than an IllegalStateException from the setCell
  that crossed the line.
• Byte size is not backstopped anywhere. MAX_BYTE_SIZE appears only in addMutation, so a
  Mutation seeded from protos can exceed 200 MB with nothing client-side objecting.

The change

Count wrapped protos in the three factories, so the counters describe the whole row rather than the
part that happened to arrive through addMutation.

Deliberately not a checkState in the factories themselves: that would make wrapping an
already-over-limit proto throw where it currently does not, which is a larger behaviour change than
restoring the guard requires. Counting never throws by itself; addMutation remains the only
place that checks.

Behaviour change

This is the point worth a reviewer's attention rather than the code. A caller that wraps protos and
then adds more mutations will now get IllegalStateException from the setCell that crosses a
limit, where today it fails later (count) or not at all (bytes). I believe that is what the limits
are for — failing fast client-side instead of sending a request that cannot succeed — but it is a
visible change and I have no objection to it waiting for a release note or a major.

Nothing else moves: no new fields, no serialization change, no API change, and readObject needs
no adjustment because numMutations and byteSize are not transient.

Verification

The full models package passes (224 tests). The two new tests are discriminating, not
characterizing — verified by removing the production change and re-running, which is the check I
care about here since the fix is a behaviour change and a test that passed either way would prove
nothing:

State	Result
with the fix	224 pass
production change reverted, tests kept	tooManyMutationsCountsWrappedProtosTest and tooLargeRequestCountsWrappedProtosTest fail (2 failures)

tooLargeRequestCountsWrappedProtosTest allocates two 100 MB values to cross the 200 MB limit,
mirroring the footprint of the existing tooLargeRequest; it runs in the same class and needed no
heap beyond what that test already implies. google-java-format reports both changed files
compliant.

Related

Independent of the other two findings in this area — the duplicate proto construction in
MutateRowsBatchingDescriptor.createResource(), and the missing serialized-size accessor on
RowMutationEntry. This PR touches the same three factories as the latter, so if both are taken
the second to merge needs a trivial rebase.

[gemini-code-assist]

Code Review

This pull request ensures that mutations wrapped from existing proto objects are correctly counted towards the mutation and byte size limits when using the fromProto and fromProtoUnsafe factory methods, and adds corresponding unit tests. The review feedback correctly identifies a potential bug where iterating over the input Iterable twice can fail or cause issues if the iterable is non-repeatable. The reviewer suggests refactoring countAllTowardsLimits to be parameterless and iterate over the internal mutations list instead.

[gemini-code-assist]

Iterating over the input protos parameter twice (once in mutation.mutations.addAll(protos) and once in countAllTowardsLimits(protos)) is inefficient and can cause bugs if protos is a non-repeatable Iterable (e.g., a one-time stream or iterator wrapper). In such cases, the second iteration will either yield no elements (leaving the counters at zero) or throw an exception.

Since mutation.mutations already contains all the added elements, we can change countAllTowardsLimits to take no arguments and iterate over this.mutations instead. This avoids double-traversal and ensures correctness for all Iterable types.

Suggested change

	/** Counts mutations wrapped from existing protos, which never reach {@link #addMutation}. */
	private void countAllTowardsLimits(Iterable<com.google.bigtable.v2.Mutation> protos) {
	for (com.google.bigtable.v2.Mutation proto : protos) {
	countTowardsLimits(proto);
	}
	}
	/** Counts mutations wrapped from existing protos, which never reach {@link #addMutation}. */
	private void countAllTowardsLimits() {
	for (com.google.bigtable.v2.Mutation proto : mutations) {
	countTowardsLimits(proto);
	}
	}

[gemini-code-assist]

Update the call to countAllTowardsLimits to use the parameterless version to avoid double-traversal of the input collection.

Suggested change

	mutation.countAllTowardsLimits(protos);
	mutation.countAllTowardsLimits();

[gemini-code-assist]

Update the call to countAllTowardsLimits to use the parameterless version to avoid double-traversal of the input collection.

Suggested change

	mutation.countAllTowardsLimits(protos);
	mutation.countAllTowardsLimits();

[gemini-code-assist]

Update the call to countAllTowardsLimits to use the parameterless version to avoid double-traversal of the input collection.

Suggested change

	mutation.countAllTowardsLimits(protos);
	mutation.countAllTowardsLimits();

[laughingman7743]

Good catch — real bug, fixed, though not quite as suggested.

The suggested replacement does not compile: mutations is an ImmutableList.Builder, which does
not implement Iterable, so it cannot be the subject of a for-each (error: for-each not applicable to expression type ... found: Builder<String>).

Instead the add and the count now happen in one pass over the input, which is a single traversal
and correct for any Iterable:

private void addAllFromProto(Iterable<com.google.bigtable.v2.Mutation> protos) {
  // One traversal: protos may be a non-repeatable Iterable.
  for (com.google.bigtable.v2.Mutation proto : protos) {
    mutations.add(proto);
    countTowardsLimits(proto);
  }
}

fromProtoUnsafeTraversesTheIterableOnceTest pins it with an Iterable that throws on a second
iterator() call; reinstating the two-pass version fails it.