feat: Support compute engine to use multi feature views as source

[HaoXuAI]

What this PR does / why we need it:

Nonbreaking changes, backward compatible.

Support multi views in source. E.g,

1. chained feature_view -> base feature_view -> data_source

base_fv = BatchFeatureView(
        name="hourly_driver_stats",
        entities=[driver],
        schema=[
            Field(name="conv_rate", dtype=Float32),
            Field(name="acc_rate", dtype=Float32),
            Field(name="avg_daily_trips", dtype=Int64),
            Field(name="driver_id", dtype=Int32),
        ],
        online=True,
        offline=True,
        source=source,
    )

chained_fv = BatchFeatureView(
        name="daily_driver_stats",
        entities=[driver],
        udf=transform_feature,
        udf_string="transform",
        schema=[
            Field(name="conv_rate", dtype=Float32),
            Field(name="driver_id", dtype=Int32),
        ],
        online=True,
        offline=True,
        source=base_fv,
        sink_source=SparkSource(
            name="daily_driver_stats_sink",
            path="/tmp/daily_driver_stats_sink",
            file_format="parquet",
            timestamp_field="event_timestamp",
            created_timestamp_column="created",
        ),
    )

2. feature_view -> [feature_view1, feature_view2]

multi_view = BatchFeatureView(
        name="multi_view",
        entities=[driver],
        schema=[
            Field(name="driver_id", dtype=Int32),
            Field(name="daily_driver_stats__conv_rate", dtype=Float32),
            Field(name="daily_driver_stats__acc_rate", dtype=Float32),
        ],
        online=True,
        offline=True,
        source=[base_fv, chained_fv],
        sink_source=SparkSource(
            name="multi_view_sink",
            path="/tmp/multi_view_sink",
            file_format="parquet",
            timestamp_field="daily_driver_stats__event_timestamp",
            created_timestamp_column="daily_driver_stats__created",
        ),
    )

Diagram:

APIs:

1. sink_source: The sink_source is used when you have multiple views as the source. This is required for materialization to understand what output you need to write to the offline store.
2. Underline default join operation: data source can be merged via either the transformation udf you specified yourself, or the default join operation. The default join operation is an inner join on each FeatureView's features, and left join on entity df.
3. Works for both BatchFeatureView and StreamFeatureView.

This unlocks the request to join multiple data sources, such as SparkSource + SnowflakeSource.
You can do with this setups:

1. Create one Snowflake FeatureView for SnowflakeSource
2. Create one Spark FeatureView for SparkSource
3. Create one FeatureView to use those FeatureView as the source.

Which issue(s) this PR fixes:

#5444 (comment)

Misc

TODO:

1. Not every feature view node needs to be recomputed. Features computed and loaded into the offline store can be skipped for materialization. This is also known as the stateful store.
2. Default join operation should be customizable.

[franciscojavierarceo]

Maybe we should just call it sink?

[HaoXuAI]

Sounds good, will update it

[franciscojavierarceo]

Some small nits but this mostly lgtm

Can you add a page on the docs before merging this PR? Would be great to share with community on how to use it.

[franciscojavierarceo]

Why not call it topological_sort?

[ntkathole]

from_proto() method recursively calls itself for each source view without any depth limit. While cycle detection exists in FeatureResolver, cycle detection only runs when you use the compute engine, but proto deserialization happens much earlier during APIs/ registry loading.

We might need to handle this in FeatureView.from_proto().

[ntkathole]

Also, do we not need to store metadata for nested feature views ? meta=None ?

[ntkathole]

We need both cycle detection and de-duplication during serialization. It may not cause issue for few feature views but if there are many feature views, it could cause slowness.

A -> [B, C]
B -> [D, E]
C -> [D, E]

When serializing FeatureViewA = FeatureViewD and FeatureViewE get serialized twice.

[HaoXuAI]

right make sense.
I don't have any meta data required for compute engine at the moment, what do you think something useful?

[ntkathole]

In __eq__ , I think we also need to compare compare source_views, else two FeatureViews with different source dependencies will be considered equal.

same for __copy__

[HaoXuAI]

good catch

[HaoXuAI]

Merged it and add new PR with following suggestions