[SPARK-58950][SQL] Materialize repeated view references as one CTE

[xumingming]

What changes were proposed in this pull request?

Introduce a new optimizer rule, ConvertViewToMaterializedCTE, that runs in
FinishAnalysis (immediately before EliminateView). When the same view is
referenced two or more times in a query, the rule groups the occurrences (by
catalog identity plus canonicalized body) and rewrites each qualifying group
into a single forceSkipInline CTERelationDef with one CTERelationRef
per occurrence. Downstream, ReplaceCTERefWithRepartition wraps each
reference site with a shuffle boundary and exchange reuse deduplicates the
computation, so the view's body is evaluated once instead of once per
reference.

Details worth noting for reviewers:

• Occurrences are grouped by (identifier, canonicalized body); the bottom-up
  rewrite matches by identifier, so views nested inside another view's body
  convert too, with definitions emitted in topological order (referenced
  definitions first).
• Views whose bodies contain internally correlated subqueries
  (e.g. t WHERE x IN (SELECT y FROM s WHERE s.k = t.k)) convert as well: a
  view body is analyzed standalone at creation, so correlated references
  always resolve inside the body and never escape the converted definition.
• The rule is gated behind a new config,
  spark.sql.optimizer.convertViewToMaterializedCTE, off by default, named
  per the configuration naming guideline in ConfigEntry.scala.
• Eligibility is conservative: the view must be deterministic and
  non-streaming, all occurrences must agree on the effective view SQL config
  and output schema, and single-reference views are skipped.

Why are the changes needed?

In ETL workloads, views are almost always costly: they encapsulate large
scans, joins, and aggregations over big base tables. When such a view is
referenced more than once in one query, the current plan inlines the view
body at every reference site, so the costly body executes once per
reference. Materializing the view once and sharing the result across all
references avoids that repeated work, for example:

CREATE VIEW serving_log_view AS
SELECT user_id, tier, COUNT(*) AS requests
FROM serving_log GROUP BY user_id, tier;

SELECT t1.user_id, t1.requests
FROM serving_log_view t1 JOIN serving_log_view t2
     ON t1.user_id = t2.user_id AND t1.tier = 'priority';

Today the whole serving_log scan and aggregation behind the view run twice.

Does this PR introduce any user-facing change?

Yes, in the form of a new, opt-in configuration:
spark.sql.optimizer.convertViewToMaterializedCTE (default false). When
enabled, repeated references to the same view are materialized as one CTE
with a shuffle boundary per reference site and exchange reuse; when disabled
(the default), the plan is unchanged.

How was this patch tested?

Added and ran unit tests in both suites:

• ConvertViewToMaterializedCTESuite (catalyst, 16 tests) - rule-level
  coverage: conversion of self-joined views, single-reference and
  non-deterministic/streaming/config-mismatch/schema-mismatch rejection,
  references inside scalar subqueries, internally correlated bodies,
  view-over-view (nested) conversion with topological definition order,
  survival of converted definitions through the InlineCTE batch, and
  idempotency.
• ConvertViewToMaterializedCTEQuerySuite (sql/core, 6 tests) - end-to-end
  coverage: results identical to the un-converted baseline, one shuffle
  boundary per reference site, non-deterministic views not converted,
  INSERT INTO from a self-referenced view, and internally correlated and
  nested views executing correctly with the config enabled.

Commands:

build/sbt -Phive 'catalyst/testOnly *ConvertViewToMaterializedCTESuite' \
  'sql/testOnly *ConvertViewToMaterializedCTEQuerySuite'
build/sbt 'catalyst/scalastyle' 'catalyst/Test/scalastyle' 'sql/scalastyle' 'sql/Test/scalastyle'

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Pi

[uros-b]

Thank you @xumingming! LGTM, adding @peter-toth for further review

[peter-toth]

I'm a bit reluctant to proceed with this PR. Spark doesn't have a cost model that we could use to estimate the benefits and costs of such materialization. That's why Spark currently materializes only multiply referenced, non-deterministic CTEs when materialization is needed for correctness.

@cloud-fan, any thoughts on this?

[xumingming]

@peter-toth Thanks for the review. I agree with you that we don’t have a cost model to check whether materialization is the right call. Here we do not do any view materialization by default, only when user enabled the switch, we will do it. In production query we have many views that should be materialized to get better performance, so we want to add this ability.