[server][common][controller][vpj] Materialized view projection and filter support

[xunyin8]

Problem Statement

Consumers of a materialized view might not be interested in every data field and update event. Having the ability to perform projection and filter could reduce footprint and improve ingestion performance for MV consumers.

Solution

Add projection and filtering support for materialized view (MV) to be more efficient about unwanted data for view consumers. Projection can be enabled by setting projection fields in the materialized view parameters. Similarly filtering can be enabled by setting filter by fields. These two features can be enabled separately or together. If enabled together the filter by fields will be included in the projecting fields automatically. Here is an example MV configuration to illustrate the ideas:

Record containing fields: {a, b, c, d, e}
Projecting fields: {b, c}
Filtering fields: {a}

The only filtering option for now is to skip if none of the filter by fields changed. The filtering is also only applied during hybrid ingestion since it doesn't make sense to have a change filter on batch push. With the above setup we will project and write all batch data to the MV ({a, b, c}). RT updates (full PUT or UPDATE) will project and write the resulting record to the MV ({a, b, c}) only if the value of field (a) is different from the old value. All DELETE events will be written to the MV (no filtering).

Code changes

In order to achieve the above behavior there are several changes:

1. Previously we've used pub sub message headers to perform forwarding to handle chunks during NR pass-through in remote regions. This strategy will not work with projection because in order for us to perform projection on batch data in remote regions, we will need the remote partition leaders to assemble the chunks during NR pass-through. We are replacing the forwarding strategy with InMemoryChunkAssembler. To ensure leaders don't resume in-between chunks we will also buffer and delay writing the chunks to drainer until we have a fully assembled record and produced it to view topic(s). The view partition header code is left untouched in VPJ to remove deployment or rollback order requirements. i.e. VPJ can get ahead of the server. If server gets ahead of VPJ that's fine too because server's new chunking support can function on its own. We can clean up everything once everything is deployed and stable (no more rollbacks).

2. Added enforcement in controller to ensure view configs are immutable. Projection schema is generated when adding a new materialized view and stored with the view config. Since there can only be one schema version per view, the znode size should be manageable with compression. If this becomes a concern we can also store it separately or generate it on the fly. We also verify the filtering by fields and projection fields to ensure they exist in latest superset or value schema and have default values.

3. Projection is performed in ComplexVeniceWriter as part of complexPut so both VPJ and leaders can use the same code for projection. Filtering is performed in MaterializedViewWriter since current offering of change filter is applicable only to hyrbid writes.

Concurrency-Specific Checks

Both reviewer and PR author to verify

• Code has no race conditions or thread safety issues.
• Proper synchronization mechanisms (e.g., synchronized, RWLock) are used where needed.
• No blocking calls inside critical sections that could lead to deadlocks or performance degradation.
• Verified thread-safe collections are used (e.g., ConcurrentHashMap, CopyOnWriteArrayList).
• Validated proper exception handling in multi-threaded code to avoid silent thread termination.

How was this PR tested?

Integration tests and will add unit tests once have some consensus on the changes.

• New unit tests added.
• New integration tests added.
• Modified or extended existing tests.
• Verified backward compatibility (if applicable).

Does this PR introduce any user-facing or breaking changes?

• No. You can skip the rest of this section.
• Yes. Clearly explain the behavior change and its impact.

[gaojieliu]

Nit,
We can remove one call of viewWriter.getValue().getViewWriterType() if we do it before the if/else.

[gaojieliu]

Can we make a static final variable in Lazy class for null Lazy: Lazy.of(() -> null)?

[gaojieliu]

You meant the filtering logic can modify the current value, right?
Is it guaranteed that it will always modify the current value, if not, I think we can do a lazy deepcopy.

[gaojieliu]

Why do we need to setup ProcessedResult here?

[gaojieliu]

We don't need to buffer it for non-projection/filtering views, right?

[gaojieliu]

Let us try to make veniceWriter final and for testing purpose, we can offer another constructor to let the user of class to pass a VeniceWriter instead of exposing a setter.

[gaojieliu]

Will it cause any issues if we always use deterministic GUID generator?

[gaojieliu]

For view with only filtering rules, we also need the ValueProvider, right?

[gaojieliu]

This might not work.
Assume:

1. Schema1 has two fields: f1, f2 and f1 has two sub fields: f11, f12.
2. Schema2 has two fields: f1, and f1 has three sub fields: f11, f12, f13 and all of them have a default value.
3. Schema1 the schema being used by `inputRecord`.
4. Schema2 is being used by `resultRecord`.

When the logic here, the resultRecord will carry f1 with f11, f12, but schema of f1 contains f11, f12 and f13, so the serialization of resultRecord will fail.
We should always keep in mind that default value only applies at de-serialization time, not serialization time.
To be safe, we need to use SerDe to get a correct resultRecord, which means we need to serialize inputRecord with inputRecord.schema into bytes and deserialize the bytes by the resultRecord.schema.

One more way is to always deserialize the record with the value schema, which is being used to generate the project schema, that means we will need to persist that value schema in view config and use it as the reader schema in SIT, and this will guarantee the simple logic in this function would work all the time, and it might be the most efficient way to handle projection/filtering since schema evolution can remove filtering fields too.

[gaojieliu]

Please check the above comment.

[lusong64]

Any reason that you wanted use LinkedList instead of List itself?

[lusong64]

The LinkedList.clone only does shallow copy. Is this what we wanted?

[lusong64]

Make it volatile? My guess is that those booleans will be read by multiple threads otherwise you won't need such a boolean.

[lusong64]

Is it possible that two threads all see the valueExtractorConfigured being false and all go into this protected section? If it is, would the following be better?

if (materializedView.isValueProviderNeeded() && valueExtractorConfigured.compareAndSet(false, true)){
....
}