fix: [branch-1.2] apply the session's Comet configs in map stages of jobs outside a SQL execution (#7302) - #7337

…a SQL execution (apache#7302)

* fix: apply the session's Comet configs in map stages of jobs outside a SQL execution

On an executor, SQLConf.get reads the task's local properties, which
Spark fills with the session's configs only for jobs that run inside a
SQL execution. A job launched from a query's RDD (Dataset.rdd with AQE
off, QueryExecution.toRdd) runs its map stages outside one, and Spark's
SQLExecutionRDD re-applies the session's configs only in the final
stage. Comet reads its settings on the executor when it builds a native
plan, so those map stages ran with every spark.comet.* setting at its
default, including the memory pool and the configs passed to native.

Capture the session's configs on the driver where Comet builds the RDDs
and native shuffle that run native plans, and make them what SQLConf.get
returns on the task thread while the plan is built, as SQLExecutionRDD
does. Tasks inside a SQL execution are unchanged.

Closes apache#7301.

* refactor: capture only Comet configs once per shuffle and let configs Spark passed win

Capture only spark.comet.* keys, and apply them only when no conf is set
on the thread and the task got none of the session's Comet configs from
Spark, so SQLExecution.withSQLConfPropagated still takes effect. Carry
the native shuffle's one captured conf from its input RDD to the writer,
apply it inside the serialized-plan getCometIterator, and drop the test
workarounds this change makes redundant.

* fix: apply the session's Comet configs to shuffle reads and test each native plan site

A Comet shuffle read with no native plan above it in its stage, such as
one feeding a Spark operator in an intermediate stage, still read Comet
defaults. Capture the configs in CometShuffledBatchRDD and apply them in
compute and computeAsShuffleBlockIterator, which makes the wrap in
CometNativeShuffleInputRDD.compute redundant, so drop it. Make that RDD's
sqlConf required so each call site captures on the driver explicitly.

Replace the single test with one per site: the native shuffle writer, a
native plan in a map stage, the native limit, the native top-K, each
checking the memory pool every native plan resolves, and a shuffle read
under spark.comet.shuffle.readBufferSize=0.

* fix: apply the captured configs to the Celeborn frame size and capture DEBUG logs in tests

The native Celeborn writer read spark.comet.shuffle.rss.maxFrameBytes in
getWriter, before the writer applies the captured configs. Carry the
configs on NativeShuffleSpec, which getWriter can reach through the
dependency, and apply them to that read as well as in the writer, in
place of CometNativeShuffleInputRDD and its iterator.

Spark's LogAppender keeps INFO and above by default, so set its threshold
to DEBUG for the memory pool line the tests read.

---------

Co-authored-by: Andy Grove <[email protected]>
(cherry picked from commit 15305af)