[FSTORE-2116] Batch inference and training data anchored on a caller-supplied spine_df

[jimdowling]

https://hopsworks.atlassian.net/browse/FSTORE-2116

get_batch_data(spine_df) returns batch inference data for timestamps that have not
happened yet, by anchoring the feature view's query on the entities and prediction times
you supply instead of on rows the root feature group has already observed. This documents it.

What this repo contributes

A new user guide, Batch data for future timestamps, and a pointer to it from the
existing batch data page.

The guide explains why a time range returns nothing for the future, then covers the
three ways to build a set of prediction times, what serving_keys may carry, how an as-of
lookup resolves and what it returns when it matches nothing, and why max_feature_age
is what makes a stale carried-forward value visible.

It states the parts that surprise people rather than leaving them to be discovered: the
returned event time is the prediction time asked for and not the event time of the
matched row, keys and event time default to included for this call where they default
to excluded for every other get_batch_data, and ambiguous key columns come back fully
qualified. Limits, engine coverage and the refused query shapes are listed.

Verification

hopsworks-docs check (strict build), markdownlint and snakeoil all pass on a clean
tree. The page is registered in nav: and the cross-link resolves by heading id.

Loadtest regression, current build

Run bfe7550e against jim-asof-inference with the out-of-cluster runner, the modified
client driving the backend and query service built from this branch and its siblings. 17
files across both drivers: test_batch, test_batch_extra_filter,
test_batch_event_time_excluded, test_pit_join_filter_consistency,
test_pit_lookback_window, test_pit_join_pushdown, test_training_data,
test_feature_group_read, and the new conformance suite.

50 passed, 2 failed, 1 skipped, 1 h 23 m.

Both failures are understood and neither is a regression.

• test_batch_event_time_excluded fails with Values list "fg0" does not have a column named "trans_freq", the same error and the same pre-existing failure proven earlier by
  rebuilding the EAR with this change removed and redeploying.
• test_asof_batch_inference on the pyspark driver passes. The job runs the client
  baked into the Spark image, which is still the released one, so the run points it at a
  wheel built from this branch (LOADTEST_CLIENT_WHEEL): the driver puts the wheel first
  on sys.path, where zipimport shadows the installed client without rebuilding the image.
  Unset, the case stays xfail. The test asserts the line the suite prints on reaching its
  end and that the driver picked the wheel up, so a job that exits 0 having skipped the
  suite cannot read as a pass.

Not covered by this run: the e2e_scale marker, the data source tests (CONNECTORS was
not set), and every workflow outside the feature store.

Also here: entry deprecated in favour of serving_keys

Filed as FSTORE-2118 and folded into this branch rather than shipped separately, so
the API lands consistent in one release instead of disagreeing with itself for one.
That ticket is closed as duplicated.

get_feature_vector, get_feature_vectors, get_inference_helper and
get_inference_helpers now accept serving_keys. All four take the same dictionary,
so leaving the inference helpers behind would have replaced one inconsistency with
another.

fv.get_feature_vector(serving_keys={"id": 1})   # preferred
fv.get_feature_vector(entry={"id": 1})          # works, warns
fv.get_feature_vector({"id": 1})                # unchanged

serving_keys takes the position entry held and entry moves to the end of the
signature, so positional callers are unaffected. Passing entry emits a
DeprecationWarning naming the replacement. Passing both is refused: they are one
argument, so two values are a caller bug rather than a preference to resolve quietly.

The resolver is a private static method rather than a module-level function, which is
not a style preference. A module-level definition in feature_view.py displaces the
FeatureView class on its generated API reference page, which breaks every
cross-reference to a FeatureView member and fails the documentation build in strict
mode. That is how it was found.

Out of scope and left alone: TrainingDataset.get_serving_vector and
get_serving_vectors, the legacy serving path superseded by feature views, and the
internal vector server plumbing, which is not public.

Also here: training data from the same rows

training_data, train_test_split, train_validation_test_split and the three
create_* methods now take spine_df as well. The observation that prompted it:
anchoring a query on a caller-supplied dataframe is what a spine group does, and the
inference spine is the late-bound form of that, so a feature view no longer has to have
been created with a spine group to build training data from a labels frame.

train_x, test_x, train_y, test_y = feature_view.train_test_split(
    test_size=0.2,
    spine_df=labels,   # keys, an event time per row, and the label
)

Two differences from a batch read. A training row is one entity at one moment rather
than an entity scored repeatedly, so a schedule is rarely what builds the frame. And
columns the view does not define are carried through to
the output untouched, which is how the label rides along; a batch read stays strict,
because inference has no labels and an unrecognised column there is worth catching.

spine_df and spine both replace the left side of the query, so passing both is
refused. An event-time window or a lookback is suppressed under a spine, because the
spine is the population and a window on top of it would drop rows the caller asked for.

A materialisation keeps its staged spine file, because its Spark job reads it after the
call returns. A sweeper reclaims those, and the ones a read failed to delete. Before
this there was no sweeper, so the warning the client logged on a failed delete described
one that did not exist.

Two defects the cluster found

Adding the training case to the conformance suite surfaced two bugs, neither reachable
from the cases that existed before.

• Spine-bound join keys were missing from the physical read. They come from the
  spine's bindings rather than from a Join, so the walk that collects needed features
  never saw them, and DuckDB reported Values list does not have a column named <key>.
  It stayed hidden while the root feature group was always pruned. A training read keeps
  the root, and the same shape was reachable from get_batch_data with a kept root whose
  keys are not selected, so this was a latent bug in the batch path too.
• The Parquet file and its declared schema disagreed for passthrough columns. The two
  were computed separately and a passthrough column fell back to the view's schema, which
  does not define it, so the file was written as a string. One function decides both now.

Naming

The dataframe parameter is spine_df, on get_batch_data and on all six training-data
methods. The frame is a spine: it replaces the left side of the query and may carry passed
features and labels as well as keys, so naming it for the keys described only part of it.

The methods whose argument is a dict of keys keep serving_keys, where that name is exact:
get_feature_vector, get_feature_vectors and the two inference helper methods. The feature
view's own serving_keys property and the feature logging argument of the same name are
unrelated and unchanged.

API consolidation

spine_df is the only spine-related argument of get_batch_data. Two parameters came
off it, both because the API was saying the same thing in more than one way.

prediction_times is gone, losslessly. The times were always expressible as a column
of the frame, and the training-data methods already required them there, so the batch path
was the odd one out. PredictionTimes remains as the thing that builds a schedule from a
cron expression, a named interval or a list, and it gains cross(). That matters: the row
order is a documented contract, entities as given and ascending in time within each, and
leaving callers to hand-roll the cross product would have made that contract theirs.

schedule = PredictionTimes.every("daily", offset="00:00", start=tomorrow, count=7)
df = fv.get_batch_data(spine_df=schedule.cross(entities, event_time="date"))

max_feature_age moved onto the feature view. It could not follow the times into the
frame, because it bounds how stale a lookup may be, which is a property of the lookup and
not of a row. Deleting it would have removed the only defence against an as-of join
carrying a stale value forward silently. As a property of the view it applies to every read
anchored on a spine, batch inference and training data alike.

fv = fs.create_feature_view(name="air_quality_fv", query=query,
                            max_feature_age=datetime.timedelta(days=1))

It is written once at creation and is read-only afterwards, because an object-only bound is
lost the moment the view is fetched again: a training run and the inference pipeline that
scores against it are separate processes, and each would have had to remember to set the
same value. That is the training/serving skew the bound exists to prevent.
fv.max_feature_age reads it back as a timedelta, or None when the view is unbounded.

It is one value for the whole view rather than a bound per feature group. A per-group map
needs a VARCHAR column holding JSON, a name to validate against the view's feature groups,
and an error code for the name that matches nothing; a single bound needs an INT of
seconds and no validation beyond positive, and no pipeline asked for the extra resolution.
The column is feature_view.max_feature_age_secs INT NULL, added by migration V100 and
nullable, so an existing view keeps the unbounded behaviour it has today. The spine carries
maxFeatureAgeSecs, and neither the query service nor the chart needed a change.

The conformance suite proves the bound reaches training data, not just batch: unbounded,
the gap day is filled by carry-forward; read through a view created with the bound, exactly
that day comes back NULL and the caller's label is untouched.

It also proves the bound survives a round trip, which the rest of the suite cannot. Every
other bounded case reads through the object that created the view, and that object carries
the bound in memory, so all of them would pass unchanged if the column were never written.
The round-trip case fetches each bounded view again, checks it is a different object,
checks the bound came back, and reads through the fetched view to confirm the value that
came back is the one the read is held to.

Verified on jim-asof-inference after applying V100 and redeploying: the suite passes, and
the rows behind it read 43200 and 86400 seconds against the two bounded views with NULL
against the unbounded one.

Third round: what the guide gained

• The accepted frame types now include a Spark DataFrame under the Spark engine, with the two
  consequences a reader needs. It is refused under the Python engine, which has no session to
  evaluate it, and it has no row order, so the positional zip-back the page shows does not
  apply to one. Join predictions back on the serving keys instead.
• A recipe for the latest feature values over every entity the feature store knows about,
  built from feature_view.get_root_fg() and feature_group.read_primary_keys() rather than
  from a list the reader maintains. The page states what it costs, the key columns of the
  whole feature group, and what it does not cover, a joined feature group keyed on something
  the root does not carry.
• max_feature_age is now a single timedelta for the whole view, read-only after creation
  and read back with feature_view.max_feature_age. The per-feature-group map, its "*"
  catch-all and the unknown-name error are gone from the page, including from the
  training-data example.

Review round: thirteen findings, applied

A review of this change against itself, after it was otherwise finished. The two that mattered
are both in other repos and are described in full on logicalclocks/hopsworks-ee#3339 and
logicalclocks/hopsworks-api#1167: a spine column name reached SQL unvalidated, and the staleness
bound was taken from the request rather than from the feature view it is stored on. Nothing in
this repo changed for either.

One consequence is worth knowing here. The spine's wire form no longer carries
maxFeatureAgeSecs; it names the feature view it anchors, by featurestore id, name and version,
and the backend reads the bound from that view's row. A client that sends the old shape is
refused with the new code 332.

The limits also remain per request: there is no cap on how many spine reads one user or project
may have in flight, so on a shared cluster a single caller can occupy the query service with
repeated large reads. Admission control is not in this change.

Verification after the round

Backend rebuilt and redeployed on jim-asof-inference, and the conformance suite run on both
drivers against it: 2 passed, 5m35s. That run is itself the check that the bound is now read
server-side, because a backend still expecting the bound on the wire would find none and answer
the bounded cases unbounded, which those cases assert against.

External review round: eleven findings, applied

An external review (Codex) of the finished change found eleven defects, six of them release
blockers, and four wider gaps. All eleven are fixed on the branches; none was answered in a
comment. The two that would have shipped wrong data:

• Materialised training dropped spine_df at two boundaries. _create_training_dataset
  accepted the frame and did not pass it on, and the Spark job that materialises a split or an
  external-sink dataset rebuilds its query from the feature view, so the spine could not reach
  it at all. Every create_* call with spine_df built the ordinary historical population while
  batch inference used the caller's rows: a training/inference population mismatch, silent. The
  spine now travels in the job configuration, validated against the view's query by the same
  resolver the read path uses (file existence and READ permission included), and the job reads
  the staged Parquet back and anchors on it. A file that is gone fails the job rather than
  building another population. Training datasets record spine_anchored (V102), so a version
  built from a spine cannot later be read or recreated from the view's own rows.
• The file check did not establish READ. exists and getFileStatus succeed with traverse
  permission on the directories alone, and the signed query-service read then ran as the
  superuser. The resolver now calls Hadoop's access(path, READ) as the requester before
  signing.

The other nine: a root filter naming a joined feature rendered inside the root's subquery and
failed to bind on both engines (predicates are now routed to the feature group they reference;
one spanning two groups is refused); event times of different types compared raw, and an integer
prediction time was parsed as nanoseconds (the spine is written in the root's event time type,
integers are epoch milliseconds, and a lookup of a different type is converted with the engine's
own function); tie-break primary keys and the event time were missing from the physical
projection; an explicit lookback was silently dropped with a spine (it now composes as a candidate
filter); Spark frames were collected before the row ceiling applied; the query service
materialised the whole spine file before checking its row count (footer first, load bounded);
Spark inferred the spine schema and refused an all-NULL key (it now registers the typed Arrow
table the Python path writes); a passthrough column could shadow a joined feature's output name;
and a dropped lookup's array or struct feature came back as an untyped NULL (each engine's own
type spelling is rendered).

The full list with dispositions is in the workspace's FSTORE-2116_review.md.

Verification after the round

Cluster jim-asof-inference. Backend rebuilt and redeployed from the branch, with V102 applied
and recorded in flyway_schema_history; the query service rolled to a rebuilt image carrying the
loader change. Conformance suite, both drivers, with a wheel built from the branch:
pyspark driver 1 passed (the branch wheel on the job's sys.path), python driver 1 passed
(4m43s). The python driver's job case needs the branch entrypoint and client inside the Spark
image, which still carries the released ones, so for the run fs_py_job_util was pointed at a
HopsFS copy of hsfs_utils.py that puts the branch wheel first on sys.path. A scripted
reproduction of that case showed the job configuration carrying spine, the job running the branch
client, a six-row frame twelve hours off every root row producing 3 + 3 rows, and the dataset
recorded as spine-anchored.

Suites: hopsworks-ee query package and Arrow Flight controller tests pass, checkstyle clean;
hopsworks-api 4985 passed, 13 skipped, plus the fifteen WINDOWED golden shapes bound against a local Spark
4.1.1; flyingduck 32 spine tests and the fifteen ASOF shapes bound against DuckDB 1.5.2; docs
strict build, markdownlint and snakeoil clean.

Not covered: the job path on a cluster whose Spark image predates this change now fails at
argument parsing instead of building the wrong population, but that refusal itself was not run on
a cluster (it needs an image with the old entrypoint and a backend with the new op).

In this repo

The user guide records that a training dataset built from spine_df needs the frame again and
refuses to run without it; that the prediction time is written in the root's event time type and
joined groups may differ; that a lookback composes with a spine; that a filter on a joined feature
restricts its candidates and one spanning two groups is refused; that max_feature_age applies to
spine-anchored reads only; and what bounds Spark's candidate cardinality.

Related pull requests

Merge in this order. The chart is the only writer of hopsworks.variables, so
the backend reads its enum defaults until the chart lands.

1. logicalclocks/hopsworks-ee#3339 and logicalclocks/hopsworks-helm#2343, together
2. logicalclocks/flyingduck#248
3. [FSTORE-2116] Batch inference and training data anchored on a caller-supplied spine_df hopsworks-api#1167
4. logicalclocks/hopsworks-front#2103, independent of logicalclocks/hopsworks-front#2106
5. logicalclocks/loadtest#1038 and [FSTORE-2116] Batch inference and training data anchored on a caller-supplied spine_df #653
6. airquality: read the forecast through the feature view featurestorebook/mlfs-book#56, the air quality example that consumes it

🤖 Generated with Claude Code

https://claude.ai/code/session_01F22onm6M6KvU686d5imQw5