SparkOfflineStore.offline_write_batch ignores table_format and writes with file_format #6955

Problem

SparkSource carries two independent format axes, and says so:

# contrib/spark_offline_store/spark_source.py:45
file_format: Optional[str] = None,
table_format: Optional[TableFormat] = None,
...
# :66
    table_format: The table metadata format (iceberg, delta, hudi, etc.).
        Optional and separate from file_format.

They are separate fields in the proto too:

// DataSource.proto
message SparkOptions {
  ...
  string file_format = 4;
  TableFormat table_format = 6;
}

offline_write_batch only consults file_format. Its path branch uses it twice — once to read the existing dataset for its schema, once to write:

# contrib/spark_offline_store/spark.py:398
df_existing = spark_session.read.format(
    feature_view.batch_source.file_format     # :400
).load(feature_view.batch_source.path)
...
df_batch.write.format(feature_view.batch_source.file_format).mode(   # :406
    "append"
).save(feature_view.batch_source.path)

So a SparkSource addressed by path whose format is carried on table_format has no correct write. With table_format set and file_format left unset — which is what the two-axis split invites — both calls receive None.

The catalog-addressed case does not reach a wrong writer, but has no writer at all:

# :414
raise NotImplementedError(
    "offline_write_batch not implemented for batch sources specified by a table"
)

Not specific to one table format

This affects anything expressed through table_format on a SparkSource: Iceberg, Delta, Hudi, and LanceFormat from #6925. read is unaffected, since the Spark reader path does consult table_format, so the two directions currently disagree about which axis is authoritative.

IcebergSource is unaffected in practice, but only because it is a separate data source class with its own branch at the top of the same function:

# :369
if isinstance(feature_view.batch_source, IcebergSource):
    _iceberg_offline_write_batch(config, feature_view, table)
    return

That branch is the precedent I would follow, but it keys off the data source class, so it does nothing for a plain SparkSource that merely has a table_format.

Why I am raising it rather than sending a patch

I do not think the fix is obvious, and the choice is a design decision rather than a bug fix:

1. Dispatch on table_format when present, file_format otherwise. Smallest change and matches what the reader already does. Needs a decision on what a source with both set means — today that is accepted silently.
2. Give each table format a dedicated branch, as IcebergSource has. Explicit, but grows the function per format and still leaves plain SparkSource + table_format unhandled unless the branch keys off the format rather than the class.
3. Reject table_format on the write path outright until a format opts in, which at least replaces a wrong writer with an error.

There is also a prior question: if table_format is authoritative for writes, should offline_write_batch keep going through read + _cast_data_frame + write.format(...).save(path) at all, or use writeTo(...) for catalog-managed formats? That would also give the catalog-addressed case at :414 a path, which the current structure cannot.

Happy to implement whichever direction maintainers prefer. I ran into this looking at what a Spark write path for Lance would need, so I have a concrete case to test against, but I would rather not guess at the intended semantics of the two axes.

Reproduction

A SparkSource(path=..., table_format=LanceFormat()) with no file_format reaches spark_session.read.format(None) at :400. Any table_format reproduces it the same way; the format itself is not what matters.