The Dask/File historical retrieval path drops an input entity row when matching source records exist but all are outside the point-in-time window. An entity with no source records is retained, so the output cardinality depends on whether an unusable record happens to exist.

This silently removes caller-supplied columns such as training labels, and can also discard valid features already joined from another feature view.

Minimal reproduction on current master (d6cc1dc), using a real Parquet source and local FeatureStore:

from datetime import timedelta
from pathlib import Path
from tempfile import TemporaryDirectory

import pandas as pd
from feast import Entity, FeatureStore, FeatureView, Field, FileSource, RepoConfig
from feast.types import Float64, ValueType

with TemporaryDirectory() as directory:
    root = Path(directory)
    source_path = root / "stats.parquet"
    pd.DataFrame({
        "driver_id": [1],
        "event_timestamp": pd.to_datetime(["2025-01-01T00:00:00Z"]),
        "value": [10.0],
    }).to_parquet(source_path)
    driver = Entity(name="driver", join_keys=["driver_id"],
                    value_type=ValueType.INT64)
    view = FeatureView(
        name="stats", entities=[driver],
        schema=[Field(name="value", dtype=Float64)],
        source=FileSource(path=str(source_path), timestamp_field="event_timestamp"),
        ttl=timedelta(days=1),
    )
    store = FeatureStore(config=RepoConfig(
        project="ttl_rows", registry=str(root / "registry.db"),
        provider="local", offline_store={"type": "file"},
        online_store={"type": "sqlite", "path": str(root / "online.db")},
    ))
    store.apply([driver, view])
    entities = pd.DataFrame({
        "driver_id": [1, 2],
        "event_timestamp": pd.to_datetime(["2025-01-03T00:00:00Z"] * 2),
        "label": ["expired", "missing"],
    })
    print(store.get_historical_features(
        entity_df=entities, features=["stats:value"]
    ).to_df())

Expected: both input rows and labels survive, with null value for both.

Actual: only driver 2 survives. Driver 1 and the expired label disappear. The same problem occurs with future-only source records, including when TTL is zero or unset.

_merge creates a left join, then _filter_ttl removes every candidate for that request. Null source timestamps are retained, but existing out-of-window records do not become null left-join results.

This is distinct from #6787: the reproduction has unique (join key, entity timestamp) pairs and no duplicate input requests.

I have a fix that nulls only the current source's ineligible columns for entity-dataframe retrieval, retaining entity columns and previously joined features. Retrieval without an entity dataframe keeps its existing snapshot filtering. On the original code, 13 of 14 new real-FeatureStore regressions fail; with the fix, 38 related tests pass (5 unrelated optional-backend tests are skipped), and Ruff plus scoped mypy pass.