[server][common][vpj] Introduce ComplexVenicePartitioner to materialized view

[xunyin8]

[server][common][vpj] Introduce ComplexVenicePartitioner to materialized view

The change will not work if record is actually large and chunked. Proper chunking support is needed and will be addressed in a separate PR.

1. Introduced ComplexVenicePartitioner which extends VenicePartitioner and offer a new API to partition by value and provide possible one-to-many partition mapping.

2. Added value provider of type Lazy to VeniceViewWriter's processRecord API to access deserialized value if needed. e.g. when a ComplexVenicePartitioner is involved.

3. MergeConflictResultWrapper and WriteComputeResultWrapper will now provide deserialized value in a best effort manner. This is useful when we already deserialized the value for a partial update operation so that the deserialized value can be provided directly to the materialized view writer.

4. Refactored VeniceWriter to expose some APIs to child class. Introduced ComplexVeniceWriter which extends VeniceWriter. Reasoning here is that the ComplexVeniceWriter will have different APIs to be used in MaterializedViewWriter and CompositeVeniceWriter to write to materialized view partition(s) and potentially involving a ComplexVenicePartitioner. Alternatively we could push common logic from VeniceWriter to AbstractVeniceWriter. However, ComplexVeniceWriter needs/shares too much common logic with VeniceWriter (chunking, DIV support, pubSubAdapter, etc...) it will make AbstractVeniceWriter too specialized and unable to offer the flexibility it needs to support something like the CompositeVeniceWriter.

5. Override putLargeValue in ComplexVeniceWriter to skip chunking and writing large messages. Once we have proper chunking support we need to be careful to not re-chunk when writing the same value to different partition in ComplexVeniceWriter.

How was this PR tested?

Added new integration test with A/A, W/C and a new test value based partitioner.
Will add new unit tests once we have some consensus on the API changes

Does this PR introduce any user-facing changes?

• No. You can skip the rest of this section.
• Yes. Make sure to explain your proposed changes and call out the behavior change.

[FelixGV]

Just some early thoughts... did not read the whole PR yet. But hopefully useful in terms of discussing the API changes.

[gaojieliu]

Code change looks good overall.
I do think we need to take care of the comment I just left, which is very tricky as it is a race condition.

[gaojieliu]

I think we need to actively lookup the previous value when there is a complex view writer:
https://github.com/linkedin/venice/blob/main/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/ActiveActiveStoreIngestionTask.java#L458

  if (hasChangeCaptureView) {
      /**
       * Since this function will update the transient cache before writing the view, and if there is
       * a change capture view writer, we need to lookup first, otherwise the transient cache will be populated
       * when writing to the view after this function.
       */
      oldValueProvider.get();
    }

I guess the condition can be delete op + complex partitioner.

[gaojieliu]

Thanks for the change!