[Sandbox] Narrow scan for sort+limit queries whose output is a subset of sort/filter columns

[rishabhmaurya]

Description

PPL queries of the shape ... | sort <col> | head <k> | fields <cols> could read every column in the table even when only a few columns are needed, when the projected (output) columns were a subset of the sort/filter columns.

The analytics engine has a late-materialization rewrite that narrows the scan to only the columns a query needs (sort keys + filter columns below the sort, plus the display columns above it). It was declining to fire whenever the set of output columns was already contained in the set of sort/filter columns, on the assumption that in that case the ordinary (non-rewritten) plan would read the same narrow set of columns anyway — so the rewrite would save nothing.

That assumption does not hold. When the sole output column is the sort key (or a subset of the sort/filter columns), the column projection is not pushed down into the scan, so the non-rewritten plan reads the full row — all columns — to answer what is effectively a one- or two-column query. Declining the rewrite in this case is a latency cliff.

Approach

• Before: fire the rewrite only when there is at least one output column that is not already a sort/filter column. Otherwise decline — and fall back to a plan that reads all columns.
• After: fire the rewrite whenever the query touches fewer columns than the full table, so the scan is narrowed to exactly the needed columns. Decline only in the genuine no-op case where the query needs every column.

This keeps the existing behavior for the common case (output columns are a superset of sort/filter columns) and additionally covers the degenerate case (output ⊆ sort/filter), where the narrowed scan now avoids materializing every column for every row. When there are no extra display columns to defer, the rewrite degenerates to "narrow scan + re-fetch the same few needed columns for the K survivors" — negligible next to the columns it stops reading for all rows.

Shard physical plans (before → after)

Captured on the data node. The key change is the scan's projection: before, it lists every column; after, it carries only the columns the query needs (the sort-key column indices drop from their position among all columns, e.g. @16, to @0).

q25 — where SearchPhrase != '' | sort EventTime | fields SearchPhrase | head 10

Before:

SortPreservingMergeExec: [EventTime@16 ASC], fetch=10
  SortExec: TopK(fetch=10), expr=[EventTime@16 ASC], preserve_partitioning=[true]
    FilterExec: SearchPhrase@74 != 
      DataSourceExec: file_groups={4 groups: [...17 parquet files...]},
        projection=[AdvEngineID, Age, BrowserCountry, BrowserLanguage, CLID, ClientEventTime,
          ClientIP, ClientTimeZone, CodeVersion, ConnectTiming, CookieEnable, CounterClass,
          CounterID, DNSTiming, DontCountHits, EventDate, EventTime, FUniqID, FetchTiming,
          FlashMajor, FlashMinor, FlashMinor2, FromTag, GoodEvent, HID, HTTPError, HasGCLID,
          HistoryLength, HitColor, IPNetworkID, Income, Interests, IsArtifical, IsDownload,
          IsEvent, IsLink, IsMobile, IsNotBounce, IsOldCounter, IsParameter, IsRefresh,
          JavaEnable, JavascriptEnable, LocalEventTime, MobilePhone, MobilePhoneModel, NetMajor,
          NetMinor, OS, OpenerName, OpenstatAdID, OpenstatCampaignID, OpenstatServiceName,
          OpenstatSourceID, OriginalURL, PageCharset, ParamCurrency, ParamCurrencyID,
          ParamOrderID, ParamPrice, Params, Referer, RefererCategoryID, RefererHash,
          RefererRegionID, RegionID, RemoteIP, ResolutionDepth, ResolutionHeight, ResolutionWidth,
          ResponseEndTiming, ResponseStartTiming, Robotness, SearchEngineID, SearchPhrase,
          SendTiming, Sex, SilverlightVersion1, SilverlightVersion2, SilverlightVersion3,
          SilverlightVersion4, SocialSourceNetworkID, SocialSourcePage, Title, TraficSourceID,
          URL, URLCategoryID, URLHash, URLRegionID, UTMCampaign, UTMContent, UTMMedium, UTMSource,
          UTMTerm, UserAgent, UserAgentMajor, UserAgentMinor, UserID, WatchID, WindowClientHeight,
          WindowClientWidth, WindowName, WithHash],
        file_type=parquet, predicate=SearchPhrase@63 !=  AND DynamicFilter [ empty ],
        pruning_predicate=SearchPhrase_null_count@2 != row_count@3 AND (SearchPhrase_min@0 !=  OR  != SearchPhrase_max@1),
        required_guarantees=[SearchPhrase not in ()]

After:

SortPreservingMergeExec: [EventTime@0 ASC], fetch=10
  SortExec: TopK(fetch=10), expr=[EventTime@0 ASC], preserve_partitioning=[true]
    <scan narrowed to [SearchPhrase, EventTime]>

q27 — where SearchPhrase != '' | sort EventTime, SearchPhrase | fields SearchPhrase | head 10

Before:

SortPreservingMergeExec: [EventTime@16 ASC, SearchPhrase@74 ASC], fetch=10
  SortExec: TopK(fetch=10), expr=[EventTime@16 ASC, SearchPhrase@74 ASC], preserve_partitioning=[true]
    FilterExec: SearchPhrase@74 != 
      DataSourceExec: file_groups={4 groups: [...17 parquet files...]},
        projection=[AdvEngineID, Age, BrowserCountry, ... (all ~100 columns, same list as q25) ... WithHash],
        file_type=parquet, predicate=SearchPhrase@63 !=  AND DynamicFilter [ empty ],
        pruning_predicate=SearchPhrase_null_count@2 != row_count@3 AND (SearchPhrase_min@0 !=  OR  != SearchPhrase_max@1),
        required_guarantees=[SearchPhrase not in ()]

After:

SortPreservingMergeExec: [EventTime@0 ASC, SearchPhrase@1 ASC], fetch=10
  SortExec: TopK(fetch=10), expr=[EventTime@0 ASC, SearchPhrase@1 ASC], preserve_partitioning=[true]
    <scan narrowed to [EventTime, SearchPhrase]>

Benchmark

ClickBench dataset (~100M rows, ~100 columns), single shard per node. Times are warm, two runs each. To isolate the effect of this change, all other query-acceleration paths were disabled for the measurement, so the numbers reflect the projection-narrowing alone.

ClickBench query	PPL	Before	After
(probe)	sort EventTime | head 10 | fields EventTime	~11.3 s	~0.23 s
q25	where SearchPhrase != '' | sort EventTime | fields SearchPhrase | head 10	~13.1 s	~0.59 s
q27	where SearchPhrase != '' | sort EventTime, SearchPhrase | fields SearchPhrase | head 10	~12.5 s	~0.60 s

The first row is a synthetic single-column probe (output column == sole sort key) — the clearest case of the degenerate shape this PR fixes. q25 and q27 are standard ClickBench queries that exhibit the same pattern (output column is a subset of the sort/filter columns).

Queries that already projected a superset of the sort/filter columns are unaffected (the rewrite already fired for them).

Testing

• LateMaterializationPlanShapeTests — the two cases that previously asserted the rewrite was declined (output = sole sort key; output ⊆ sort+filter) are converted to assert the narrowed-scan plan. Full suite (26 tests) passes.
• spotlessJavaCheck clean.

Check List

• Functionality includes unit tests.
• Commits are signed per the DCO using --signoff.

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ No major issues detected

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Align scan column count with physical fields totalScanCols counts all fields in the scan's row type, but neededPhysicalFields likely excludes synthetic/system columns (e.g., _id, _score) that are present in the scan's row type. This mismatch can cause narrowsScan to be incorrectly true (or miss the no-op case), firing QTF when narrowing actually adds no benefit. Compare against the count of physical fields in the scan consistent with how belowAnchorPhysicalFields is computed. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchLateMaterializationRewriter.java [156-158] Set<String> neededPhysicalFields = new HashSet<>(belowAnchorPhysicalFields); neededPhysicalFields.addAll(aboveAnchorPhysicalFields); -int totalScanCols = belowChain.scan.getRowType().getFieldCount(); +int totalScanCols = (int) belowChain.scan.getRowType().getFieldList().stream() + .filter(f -> !f.getName().startsWith("_")) + .count(); Suggestion importance[1-10]: 3 __ Why: The suggestion raises a plausible concern about synthetic columns affecting totalScanCols, but it's speculative without evidence from the code that such columns exist in the scan's row type. The proposed fix using a _ prefix filter is a guess and may not match the actual convention.	Low

[expani]

Thanks for taking this up @rishabhmaurya

Can you share the Datafusion Physical Plans at the Shard before and after your changes for Q25 and Q27 ?

[rishabhmaurya]

@expani updated the PR with the plan

[github-actions]

✅ Gradle check result for d69dee4: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.34%. Comparing base (64d6b50) to head (d69dee4).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #22294      +/-   ##
============================================
- Coverage     73.37%   73.34%   -0.04%     
+ Complexity    76067    76016      -51     
============================================
  Files          6075     6075              
  Lines        345368   345368              
  Branches      49718    49718              
============================================
- Hits         253424   253316     -108     
- Misses        71769    71845      +76     
- Partials      20175    20207      +32     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.