Refactor RowSink into RowSink and TupleBuffer
Cap the number of rows per shard
Bucketize shards based on bucket_key