[SPARK-49193][SQL] Improve the performance of RowSetUtils.toColumnBasedSet

[wangyum]

What changes were proposed in this pull request?

Replace while loop with foreach in RowSetUtils.toTColumn.

Why are the changes needed?

Improve the performance of RowSetUtils.toColumnBasedSet:

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Manual test.

import org.apache.hive.service.rpc.thrift.TProtocolVersion
import org.apache.spark.sql.execution.HiveResult

val df = spark.sql("select id, cast(id as string), cast(id as timestamp) from range(200000)")
val dataTypes = df.schema.fields.map(_.dataType)
val rows = df.collect().toList
val start1 = System.currentTimeMillis()
RowSetUtils.toTRowSet(1, rows, dataTypes, TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V11, HiveResult.getTimeFormatters)
val start2 = System.currentTimeMillis()
RowSetUtils.toTRowSet(1, rows, dataTypes, TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V5, HiveResult.getTimeFormatters)
val start3 = System.currentTimeMillis()
println(s"toColumnBasedSet time: ${start2 - start1}, toRowBasedSet time: ${start3 - start2}")

Before this PR:

toColumnBasedSet time: 17307, toRowBasedSet time: 71

After this PR:

toColumnBasedSet time: 128, toRowBasedSet time: 70

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

No.

[HyukjinKwon]

Hmmm generally while is faster than foreach (https://github.com/databricks/scala-style-guide?tab=readme-ov-file#traversal-and-zipwithindex)

[wangyum]

It seems the Array's performance is much better than Seq.

toColumnBasedSet time: 138, toRowBasedSet time: 73

[yaooqinn]

LGTM

[wangyum]

Another test:

import org.apache.hive.service.rpc.thrift.TProtocolVersion
import org.apache.spark.sql.execution.HiveResult

val df = spark.sql("select id, cast(id as string), cast(id as timestamp),  cast(id as decimal(18, 0)) from range(20000000)")
val dataTypes = df.schema.fields.map(_.dataType)
val rows = df.collect().toList
val start1 = System.currentTimeMillis()
RowSetUtils.toTRowSet(1, rows, dataTypes,
  TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V11, HiveResult.getTimeFormatters)
val start2 = System.currentTimeMillis()
RowSetUtils.toTRowSet(1, rows, dataTypes,
  TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V5, HiveResult.getTimeFormatters)
val start3 = System.currentTimeMillis()
println(s"toColumnBasedSet time: ${start2 - start1}, toRowBasedSet time: ${start3 - start2}")

Result:

toColumnBasedSet time: 40678, toRowBasedSet time: 90844

[HyukjinKwon]

The docmumentation build, you might need to sync your branch and master branch to the latest.

Merged to master.

[yaooqinn]

I backported this to branch-3.5 and branch-3.4