[BUG] Incorrect queryTime when running NDS

[amahussein]

Describe the bug

I noticed that the queryTime reported by the benchmark can be completely off.
For example, NDS query32 was reported to run for 8 seconds while its actual time was 1.07 seconds. The 1.07 seconds matches the SHS UI and the rapids-tools report.

The root reason is that we use the current time upon return to the python code. This is not reliable way of measuring the actual query time.

spark-rapids-benchmarks/utils/python_benchmark_reporter/PysparkBenchReport.py

Lines 101 to 103 in 2ad6b99

	start_time = int(time.time() * 1000)
	fn(*args)
	end_time = int(time.time() * 1000)

We should instead extract the query execution time from an eventlistener, or from the evntlog itself.

[mattahrens]

Consideration: NDS benchmark can report both user time (existing methodology) and Spark runtime (from eventlistener).

We'd also want to investigate what is causing the gap between the user time and Spark runtime for specific queries where this has been observed.

[amahussein]

The discrepancy between the Python-measured queryTime and the Spark event log duration stems from multiple sources of overhead that are included in the Python timing but NOT in the Spark SQL execution time.

1. What the Python Code Measures

start_time = int(time.time() * 1000)
fn(*args)  # calls run_one_query()
end_time = int(time.time() * 1000)

The fn(*args) call to run_one_query() includes:

spark-rapids-benchmarks/nds/nds_power.py

Lines 300 to 324 in f320705

	def run_one_query(spark_session,
	profiler,
	query,
	query_name,
	output_path,
	output_format,
	save_plan_path,
	plan_types,
	skip_execution):
	with profiler(query_name=query_name):
	print(f"Running query {query_name}")
	df = spark_session.sql(query)
	if not skip_execution:
	if not output_path:
	df.collect()
	else:
	ensure_valid_column_names(df).write.format(output_format).mode('overwrite').save(
	output_path + '/' + query_name)
	if save_plan_path:
	os.makedirs(save_plan_path, exist_ok=True)
	explain_str = spark_session._jvm.PythonSQLUtils.explainString(df._jdf.queryExecution(), 'extended')
	plans = parse_explain_str(explain_str)
	for plan_type in plan_types:
	with open(save_plan_path + '/' + query_name + "." + plan_type, 'w') as f:
	f.write(plans[plan_type])

Overhead Source	Enabled	Description	Impact
Profiler hooks	No*	External script execution via subprocess.run() for start/stop profiling	Can be significant (seconds)
spark_session.sql(query)	Yes	Query parsing and logical plan creation (before execution starts)	Usually small
Plan extraction	No*	PythonSQLUtils.explainString(df._jdf.queryExecution(), 'extended')	can be expensive for complex plans
File I/O	No*	os.makedirs() and writing plan files to disk	Moderate
PySpark serialization	Yes	Data transfer from JVM to Python for df.collect()	Depends on result size
Python-JVM bridge	Yes	Py4J communication overhead	Usually small

*: neither --save_plan_path nor --profiling_hook are passed by default. So those overheads are not applicable.
Given this, the remaining sources of the 8s vs 1.07s discrepancy are:

There are 2 major sources that consume times here:

The Python timing starts before spark_session.sql(query) is called. This call does:

Query Parsing, Analysis & Optimization

1. SQL parsing
2. Query analysis (resolving tables, columns)
3. Catalyst optimization (logical plan optimization)
4. Physical plan generation
   These steps happen before the SparkListenerSQLExecutionStart event is fired. For complex queries like query32, this optimization phase could take several seconds.

Py4J and Result Transfer

After Spark execution completes (SparkListenerSQLExecutionEnd), there's overhead for:

1. Py4J marshaling results from JVM to Python
2. Python processing the collected results

---

From the eventlog, I cannot see the kind of gap 7+ seconds which could indicate:

1. YARN resource allocation delays
2. Executor startup latency
3. GC pauses
4. Some CSP specific issues. For example, if this happens more frequently on a specific CSP.