feat: Support table format: Iceberg, Delta, and Hudi

[HaoXuAI]

What this PR does / why we need it:

examples:

  from feast.table_format import IcebergFormat
  from feast.infra.offline_stores.contrib.spark_offline_store.spark_source import SparkSource

  # Create Iceberg format with snapshot option
  iceberg_format = IcebergFormat()
  iceberg_format.set_property("snapshot-id", "your_snapshot_id")

  spark_source = SparkSource(
      name="iceberg_features",
      path="my_catalog.my_database.my_table",  # Catalog table name
      table_format=iceberg_format
  )

  Generated Spark code:
  spark.read.format("iceberg").option("snapshot-id", "your_snapshot_id").load("my_catalog.my_database.my_table")

  ✅ Delta Example:

  from feast.table_format import DeltaFormat

  delta_format = DeltaFormat()
  delta_format.set_property("versionAsOf", "1")

  spark_source = SparkSource(
      name="delta_features",
      path="s3://bucket/delta-table",
      table_format=delta_format
  )

  Generated Spark code:
  spark.read.format("delta").option("versionAsOf", "1").load("s3://bucket/delta-table")

  ✅ Hudi Example:

  from feast.table_format import HudiFormat

  hudi_format = HudiFormat()
  hudi_format.set_property("hoodie.datasource.query.type", "snapshot")
  hudi_format.set_property("as.of.instant", "20231010120000")

  spark_source = SparkSource(
      name="hudi_features",
      path="s3://bucket/hudi-table",
      table_format=hudi_format
  )

  Generated Spark code:
  spark.read.format("hudi")\
    .option("hoodie.datasource.query.type", "snapshot")\
    .option("as.of.instant", "20231010120000")\
    .load("s3://bucket/hudi-table")

Which issue(s) this PR fixes:

Misc

[HaoXuAI]

TODO, create TableFormat proto, consolidate with FileFormat proto

[tokoko]

+1 on the inclusion of all 3 formats. Still I think we might be able to better design data-source side such that data source definitions don't tie the sources to specific offline stores. For example right now FileSource already supports specifying delta as a format... I don't like that design either because delta and especially iceberg might require more configuration that plain files, but the upside in that design was that Source class wasn't tied to a single offline store and it is actually usable from both dask and duckdb offline stores. Hiding formats behind SparkSource will mean that when a user stores a data source instance in the registry only spark offline store can be used to retrieve the data.

I think we can have best of both worlds if we instead go for adding all these formats as separate independent data sources (DeltaSource, IcebergSource, HudiSource) and then simply teach spark engine (and in the future others) how to read it.

[franciscojavierarceo]

why not consolidate now?

[franciscojavierarceo]

I think we can have best of both worlds if we instead go for adding all these formats as separate independent data sources (DeltaSource, IcebergSource, HudiSource) and then simply teach spark engine (and in the future others) how to read it.

+1

[HaoXuAI]

@franciscojavierarceo @tokoko consolidation with FilleFormat and new data sources could break the backward compatibility, so I want to do it pace by pace.

[franciscojavierarceo]

That makes sense

[tokoko]

@HaoXuAI Why would new data sources break backwards compatibility though?

[HaoXuAI]

@HaoXuAI Why would new data sources break backwards compatibility though?

There will be some proto changes, no 100% sure if there will be API changes exposed to users but I think might be the case

[HaoXuAI]

@franciscojavierarceo @ntkathole mind take a look

[franciscojavierarceo]

@HaoXuAI i don't see use actually using or testing Spark Table, Iceberg, or Hudi format's outside of our definitions, can you add that?

Can you also add documentation that these formats are now supported?

Otherwise lgtm.

[HaoXuAI]

@HaoXuAI i don't see use actually using or testing Spark Table, Iceberg, or Hudi format's outside of our definitions, can you add that?

Can you also add documentation that these formats are now supported?

Otherwise lgtm.

[HaoXuAI]

@HaoXuAI i don't see use actually using or testing Spark Table, Iceberg, or Hudi format's outside of our definitions, can you add that?
Can you also add documentation that these formats are now supported?
Otherwise lgtm.

Gonna update to add the TableFormat proto in the next PR, after that I'll add the docs. And I think the test will need to be changed as well.