feat: skip large rows

[sarthakbhutani]

Tasks remaining -

• make changes in the read request API which skips for large rows/internally calls readLargeRowsCallable()
• expose large rows rowkeys in sidechannel/dlq/some other method - which can be exposed to client

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:

• Make sure to open an issue as a bug/issue before writing your code! That way we can discuss the change, evaluate designs, and agree on the general idea
• Ensure the tests and linter pass
• Code coverage does not decrease (if any source code was changed)
• Appropriate docs were updated (if necessary)
• Rollback plan is reviewed and LGTMed
• All new data plane features have a completed end to end testing plan

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the samples format.

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[mutianf]

I haven't looked at RowSetUtil or any of the tests but left some comments you can work on first.

[mutianf]

Since you're modifying the ServerStreamingAttemptCallable directly we can remove this class.

[mutianf]

I think this has no effect because we create a new strategy here: /~https://github.com/googleapis/sdk-platform-java/blob/main/gax-java/gax/src/main/java/com/google/api/gax/rpc/RetryingServerStreamingCallable.java#L80 (which I didn't notice previously).

[sarthakbhutani]

correct!! I believe - we create new resumption strategy, so that parallel requests have their own state tracking via new resumption strategy objects per request

[sarthakbhutani]

have resolved your comments - which are implemented in code.

unresolved comments - have not done - require a discussion - can discuss offline?

[sarthakbhutani]

this method can be implemented later. have added javadoc for the same

[sarthakbhutani]

correct!! I believe - we create new resumption strategy, so that parallel requests have their own state tracking via new resumption strategy objects per request

[sarthakbhutani]

testing this

[sarthakbhutani]

[duplicate comment]
[can discuss offline]

it is required.
it goes to the convertableExceptionCallable layer, which converts the FailedPreCondition Exception to ApiException & set the exception as ```retryable:true``

which then gets thrown to the ResumptionStrategy layer. Hence, this is required.

I have confirmed this in testing/debugging as well.

[sarthakbhutani]

added comments

[mutianf]

I don't think we need to create a base callable for this, so you can probably just extract everything from the other base callable and put it in this method. Otherwise they're all called base callables and it gets a bit confusing.

[mutianf]

I don't think you need to pass in the resumption strategy here because it's already overriden in the settings in the other callable (line 580). So this variable is not doing anything.

[mutianf]

Suggested change

	if (resumptionStrategy.getClass() == LargeReadRowsResumptionStrategy.class) {
	if (resumptionStrategy instanceof LargeReadRowsResumptionStrategy) {

[mutianf]

This still needs to be updated.

[sarthakbhutani]

have made the changes

[sarthakbhutani]

have made the changes as you suggested.
Open question - is there a reason - why we didn't do this earlier. thinking of any edge cases that may fail.

[generated-files-bot]

Warning: This pull request is touching the following templated files:

• .kokoro/presubmit/graalvm-native.cfg
• samples/snapshot/pom.xml

[mutianf]

This still needs to be updated.

[mutianf]

You can simplify this logic:

    private static List<RowRange> splitOnLargeRowKey(RowRange range, ByteString largeRowKey) {
        List<RowRange> rowRanges = new ArrayList<>();

        ByteString startKey = StartPoint.extract(range).value;
        ByteString endKey = EndPoint.extract(range).value;

        // if end key is on the left of large row key, don't split
        if (ByteStringComparator.INSTANCE.compare(endKey, largeRowKey) < 0) {
            rowRanges.add(range);
            return rowRanges;
        }

        // if start key is on the right of the large row key, don't split
        if (ByteStringComparator.INSTANCE.compare(startKey, largeRowKey) > 0) {
            rowRanges.add(range);
            return rowRanges;
        }

        // if start key is on the left of the large row key, set the end key to be large row key open
        if (ByteStringComparator.INSTANCE.compare(startKey, largeRowKey) < 0) {
            RowRange beforeSplit = range
                    .toBuilder()
                    .setEndKeyOpen(largeRowKey)
                    .build();
            rowRanges.add(beforeSplit);
        }

        // if the end key is on the right of the large row key, set the start key to be large row key open
        if (ByteStringComparator.INSTANCE.compare(endKey, largeRowKey) > 0) {
            RowRange afterSplit = range
                    .toBuilder()
                    .setStartKeyOpen(largeRowKey)
                    .build();
            rowRanges.add(afterSplit);
        }
       return rowRanges;
    }

This should cover all edge cases.

[sarthakbhutani]

have done. testing.

[mutianf]

I think you should also resume from the last seen row key.

[mutianf]

I think you want this:

    public static RowSet eraseLargeRow(RowSet rowSet, ByteString lastSeenRowKey, ByteString largeRowKey, boolean fromStart) {
        // first, remove everything we've already read from the RowSet
        RowSet remaining = erase(rowSet, lastSeenRowKey, fromStart);

        // return null if we've read everything
        if (remaining == null) {
            return null;
        }

        // second, remove the large row key from the remaining RowSet
        RowSet.Builder newRowSet = RowSet.newBuilder();

        // remove large row key from point reads
        remaining.getRowKeysList().stream().filter(k -> !k.equals(largeRowKey)).forEach(newRowSet::addRowKeys);

        // remove large row key from row ranges
        for (RowRange range : remaining.getRowRangesList()) {
            List<RowRange> afterSplit = splitOnLargeRowKey(range, largeRowKey);
            if (!afterSplit.isEmpty()) {
                afterSplit.forEach(newRowSet::addRowRanges);
            }
        }
        
        if (newRowSet.getRowKeysList().isEmpty() && newRowSet.getRowRangesList().isEmpty()) {
            return null;
        }

        return newRowSet.build();
    }

splitOnLargeRowKey is in my other comment.

And in your resumption strategy, you wouldn't need to keep the previous request anymore, because that'll be removed by the RowSet remaining = erase(rowSet, lastSeenRowKey, fromStart);

[sarthakbhutani]

we needed to keep the prev request, because, for the use case of multiple large row keys together.
if we dont keep the prev request, if the prev request failed because of large-row and the next request also failed because of large-row, in this request, we will remove the 2nd large-row key & the prev one would be there, this would keep failing

ex -
request [r1,r4]
r1 - success key
r2 - large-row key
r3 - large-row key
r4 - large-row key

original request -> [r1,r4]
r1 read, r2 fails -> req becomes (r1,r2),(r2,r4]
r3 fails, req becomes -> (r1,r3), (r3,r4] -> which will now fail for r2 again (if the prev request or prev failed row keys are not cached)

[mutianf]

Please remove this file

[sarthakbhutani]

add comments

[sarthakbhutani]

have done. testing.

[mutianf]

/gcbrun