← 输入框|github.com|刷新
Skip to content

feat: Push column projection and time-range filters down to Iceberg scans in the DuckDB and Spark offline stores #7019

Open

Description

@haoxu0

Is your feature request related to a problem? Please describe.

Follow-up to #6499 / #7005 (Polaris support, merged in #7006). Reads from an IcebergSource are always a full-table scan:

  • DuckDB offline store: _read_iceberg_catalog_source does table.scan().to_arrow() and wraps the result in ibis.memtable(...). The column selection and time-range filters that pull_latest_from_table_or_query_ibis, pull_all_from_table_or_query_ibis and get_historical_features_ibis apply afterwards are evaluated by DuckDB on the already-materialized table, so they never reach the Iceberg scan.
  • Spark offline store: _register_iceberg_source_as_temp_view does scan().to_arrow() and createDataFrame(...), so every column of the table is loaded into the driver regardless of what the feature view uses.

For wide or large tables this means reading all columns and all partitions/files just to use a few columns in a bounded time window, which also defeats Iceberg's column projection and partition/file pruning.

The data needed to avoid this is already known where the read happens, but the reader contract is data_source_reader(data_source, repo_path) -> Table, which has no way to carry it:

  • pull_latest_* / pull_all_*: the exact columns (join_keys + features + timestamp fields) and start_date / end_date.
  • get_historical_features_ibis.read_fv: the join keys, requested feature columns and timestamp columns of each feature view, and the entity dataframe's event-timestamp range (a feature row can only matter if timestamp <= max(entity ts), and, with a TTL, timestamp >= min(entity ts) - ttl).

Describe the solution you'd like

  • Add a small DataSourceReadHints dataclass (columns, timestamp_field, start, end) and an optional hints-aware reader alongside the existing data_source_reader in the ibis helpers. Existing readers (MSSQL, Oracle, file-based) keep working unchanged; offline stores that supply the hints-aware reader get pushdown.
  • DuckDB: pass the hints to _read_iceberg_catalog_source, which calls PyIceberg scan(selected_fields=..., row_filter=...).
    • Translate Feast-side column names back to source names through field_mapping, and only select columns that exist in the table schema.
    • Only build a time filter when the timestamp column is an Iceberg timestamp / timestamptz (using the matching tz-aware / naive UTC literal). Otherwise skip it.
    • The filter is a pure optimization: the existing ibis filters are still applied afterwards, so results must be identical with and without pushdown.
  • Spark: prune the registered temp view to the columns the feature view needs (and apply the time bounds where the entity range is available).
  • Unit tests against a real PyIceberg SqlCatalog (not only mocks) that assert both the scan arguments and that results match the unpruned read; docs note.

Describe alternatives you've considered

  • Predicate pushdown on entity keys (join_key IN (...)): potentially large win for small entity dataframes, but riskier (large IN lists, composite keys, type coercion). Left out of the first iteration.
  • Lazy scans via DuckDB's Iceberg extension instead of Arrow materialization: bigger change to how the offline store reads, and would not cover Spark.
  • Changing the data_source_reader signature for all readers: breaking for the MSSQL/Oracle stores that share these helpers.

Additional context

Related: #6499, #7005, #7006. I plan to open a PR for this.

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions