[KYUUBI #7539] [Spark]  Remove rebalance shuffle before writing at AQE time when not worthwhile

## Why are the changes needed?

`RebalanceBeforeWriting` injects a `RebalancePartitions` operator before writes (physically a
`ShuffleExchangeExec` with `shuffleOrigin = REBALANCE_PARTITIONS_BY_NONE`) to merge small files and
avoid the small-file problem. However, in some cases this rebalance shuffle is pure overhead: once
the upstream data is materialized and its real size is known, the shuffle brings no benefit and only
adds an extra round of shuffle read/write.

This PR adds an AQE runtime optimizer rule `RemoveRebalanceShuffle` that, at AQE time (when the
upstream query stage is materialized and real statistics are available), decides whether such a
not-worthwhile rebalance shuffle can be removed, saving one shuffle.

## What changes were proposed in this pull request?

A new rule `RemoveRebalanceShuffle`, injected into the `AQEOptimizer` via
`injectRuntimeOptimizerRule`, runs on the **logical plan**: it replaces `RebalancePartitions` with
its child, and the physical planner then re-plans without the shuffle.

**Scope** — only the rebalance feeding the final write is considered: starting from `WriteFiles`,
the rule walks down through at most `Project` and local `Sort`; any other operator in between leaves
the rebalance untouched.

**Removal conditions** (all must hold):

1. `spark.sql.adaptive.removeRebalanceShuffle.enabled` is on;
2. the `RebalancePartitions` has no partition expressions (i.e. `REBALANCE_PARTITIONS_BY_NONE`);
3. its `optAdvisoryPartitionSize` is present and **larger than**
   `spark.sql.adaptive.advisoryPartitionSizeInBytes` (a smaller advisory size sugge
   intentionally wants finer partitions, so we respect that).

**Size decision** — the rebalance input is split into groups of sub-plans whose leaves are all
materialized `LogicalQueryStage` exchange stages (a `Union` yields one group per br
measured by its runtime `sizeInBytes`. The shuffle is removed only when every group is removable,
where a group is removable under either:

- **Large data**: the input has no data-reducing operator and the group's represent
  `max(maxStageSize, groupSum / 2)` is larger than
  `spark.sql.shuffle.partitions * removeRebalanceShuffle.smallPartitionSize`; or
- **Small data**: the input has no data-expanding operator and the representative size is smaller
  than `advisoryPartitionSizeInBytes * removeRebalanceShuffle.tolerableSmallFileNum

Operator classification (logical operators):

- reducing: `Aggregate` / `Filter` / `Sample` / `Offset` / `GlobalLimit` / `LocalLi
  `WindowGroupLimit` / `Inner` join / semi-anti join;
- expanding: `Generate` / `Expand` / non-semi-anti `Join`.

**New configs** (`KyuubiSQLConf`, version 1.12.0):

| Config | Default |
|---|---|
| `spark.sql.adaptive.removeRebalanceShuffle.enabled` | `false` |
| `spark.sql.adaptive.removeRebalanceShuffle.smallPartitionSize` | `96MB` |
| `spark.sql.adaptive.removeRebalanceShuffle.tolerableSmallFileNum` | `5` |

Modules changed (identical across 3.5 / 4.0 / 4.1):

- Add `RemoveRebalanceShuffle.scala` and `RemoveRebalanceShuffleSuite.scala`;
- Add 3 configs in `KyuubiSQLConf.scala`;
- Register the rule in `KyuubiSparkSQLExtension.scala`;
- Document the feature and configs in `docs/extensions/engines/spark/rules.md`.

## How was this patch tested?

Added `RemoveRebalanceShuffleSuite` (3.5 / 4.0 / 4.1), which asserts on the presence of
`REBALANCE_PARTITIONS_BY_NONE` shuffle in the executed plan, covering:

- semi join (reducing), small data → removed;
- outer join (expanding, non-reducing), large data with tiny / huge `smallPartitionSize` → removed /
  kept respectively;
- rebalance advisory size ≤ session advisory size → kept;
- inner join (both reducing and expanding) → kept;
- semi join with `tolerableSmallFileNum=0` → kept;
- switch off → kept.

All pass on Spark 3.5:

```
build/mvn -Pspark-3.5 -pl extensions/spark/kyuubi-extension-spark-3-5 -am test \
  -Dtest=none -DwildcardSuites=org.apache.spark.sql.RemoveRebalanceShuffleSuite
```

No regression in `RebalanceBeforeWritingSuite`. Formatted with `spotless:apply` across all Spark
profiles.
### Was this patch authored or co-authored using generative AI tooling?

co-authored: Claude Opus 4.8

Closes #7539 from ulysses-you/skipshuffle.

Closes #7539

9cacd4b3a [Xiduo You] style
73e7c3faf [Xiduo You] InferRebalanceAndSortOrders
c21e100a2 [Xiduo You] coalesce
900bc9de1 [Xiduo You] docs
a74d9bc58 [Xiduo You] fix
0c19d2338 [Xiduo You] test
8c23eecbb [Xiduo You] Add RemoveRebalanceShuffle rule to remove unnecessary  rebalance

Authored-by: Xiduo You <ulyssesyou@apache.org>
Signed-off-by: Xiduo You <ulyssesyou@apache.org>
13 files changed
tree: d51aa67dba948ae87ab07abd8b2c55ce3f467399
  1. .github/
  2. .idea/
  3. bin/
  4. build/
  5. charts/
  6. conf/
  7. dev/
  8. docker/
  9. docs/
  10. extensions/
  11. externals/
  12. grafana/
  13. integration-tests/
  14. kyuubi-assembly/
  15. kyuubi-common/
  16. kyuubi-ctl/
  17. kyuubi-events/
  18. kyuubi-ha/
  19. kyuubi-hive-beeline/
  20. kyuubi-hive-jdbc/
  21. kyuubi-hive-jdbc-shaded/
  22. kyuubi-metrics/
  23. kyuubi-rest-client/
  24. kyuubi-server/
  25. kyuubi-util/
  26. kyuubi-util-scala/
  27. kyuubi-zookeeper/
  28. licenses/
  29. licenses-binary/
  30. python/
  31. .asf.yaml
  32. .dockerignore
  33. .gitattributes
  34. .gitignore
  35. .rat-excludes
  36. .readthedocs.yaml
  37. .scalafmt.conf
  38. AGENTS.md
  39. CLAUDE.md
  40. codecov.yml
  41. CONTRIBUTING.md
  42. LICENSE
  43. LICENSE-binary
  44. NOTICE
  45. NOTICE-binary
  46. pom.xml
  47. README.md
  48. scalastyle-config.xml
README.md

Apache Kyuubi

Apache Kyuubi™ is a distributed and multi-tenant gateway to provide serverless SQL on data warehouses and lakehouses.

What is Kyuubi?

Kyuubi provides a pure SQL gateway through Thrift JDBC/ODBC interface for end-users to manipulate large-scale data with pre-programmed and extensible Spark SQL engines. This “out-of-the-box” model minimizes the barriers and costs for end-users to use Spark at the client side. At the server-side, Kyuubi server and engines' multi-tenant architecture provides the administrators a way to achieve computing resource isolation, data security, high availability, high client concurrency, etc.

  • [x] A HiveServer2-like API
  • [x] Multi-tenant Spark Support
  • [x] Running Spark in a serverless way

Target Users

Kyuubi's goal is to make it easy and efficient for anyone to use Spark(maybe other engines soon) and facilitate users to handle big data like ordinary data. Here, anyone means that users do not need to have a Spark technical background but a human language, SQL only. Sometimes, SQL skills are unnecessary when integrating Kyuubi with Apache Superset, which supports rich visualizations and dashboards.

In typical big data production environments with Kyuubi, there should be system administrators and end-users.

  • System administrators: A small group consists of Spark experts responsible for Kyuubi deployment, configuration, and tuning.
  • End-users: Focus on business data of their own, not where it stores, how it computes.

Additionally, the Kyuubi community will continuously optimize the whole system with various features, such as History-Based Optimizer, Auto-tuning, Materialized View, SQL Dialects, Functions, etc.

Usage scenarios

Port workloads from HiveServer2 to Spark SQL

In typical big data production environments, especially secured ones, all bundled services manage access control lists to restricting access to authorized users. For example, Hadoop YARN divides compute resources into queues. With Queue ACLs, it can identify and control which users/groups can take actions on particular queues. Similarly, HDFS ACLs control access of HDFS files by providing a way to set different permissions for specific users/groups.

Apache Spark is a unified analytics engine for large-scale data processing. It provides a Distributed SQL Engine, a.k.a, the Spark Thrift Server(STS), designed to be seamlessly compatible with HiveServer2 and get even better performance.

HiveServer2 can identify and authenticate a caller, and then if the caller also has permissions for the YARN queue and HDFS files, it succeeds. Otherwise, it fails. However, on the one hand, STS is a single Spark application. The user and queue to which STS belongs are uniquely determined at startup. Consequently, STS cannot leverage cluster managers such as YARN and Kubernetes for resource isolation and sharing or control the access for callers by the single user inside the whole system. On the other hand, the Thrift Server is coupled in the Spark driver's JVM process. This coupled architecture puts a high risk on server stability and makes it unable to handle high client concurrency or apply high availability such as load balancing as it is stateful.

Kyuubi extends the use of STS in a multi-tenant model based on a unified interface and relies on the concept of multi-tenancy to interact with cluster managers to finally gain the ability of resources sharing/isolation and data security. The loosely coupled architecture of the Kyuubi server and engine dramatically improves the client concurrency and service stability of the service itself.

DataLake/Lakehouse Support

The vision of Kyuubi is to unify the portal and become an easy-to-use data lake management platform. Different kinds of workloads, such as ETL processing and BI analytics, can be supported by one platform, using one copy of data, with one SQL interface.

  • Logical View support via Kyuubi DataLake Metadata APIs
  • Multiple Catalogs support
  • SQL Standard Authorization support for DataLake(coming)

Cloud Native Support

Kyuubi can deploy its engines on different kinds of Cluster Managers, such as, Hadoop YARN, Kubernetes, etc.

The Kyuubi Ecosystem(present and future)

The figure below shows our vision for the Kyuubi Ecosystem. Some of them have been realized, some in development, and others would not be possible without your help.

Online Documentation

Quick Start

Ready? Getting Started with Kyuubi.

Contributing

Project & Community Status

Aside

The project took its name from a character of a popular Japanese manga - Naruto. The character is named Kyuubi Kitsune/Kurama, which is a nine-tailed fox in mythology. Kyuubi spread the power and spirit of fire, which is used here to represent the powerful Apache Spark. Its nine tails stand for end-to-end multi-tenancy support of this project.