Optimize sample perf

[res-life]

This fixes #4096

Samples data by GPU JNI to improve the performance.
Added "spark.rapids.sql.fast.sample" to switch.

[res-life]

The bottlenecks are XORShiftRandom.sample and ArrayBuffer append.
	It's hard to optimize the XORShiftRandom.sample. XORShiftRandom is serialized algorithm(current value depends on privious value), can't run paralleled.
	For ArrayBuffer append, it's also hard to avoid, we should collect the sampled row numbers.
Optimization
	Add pure GPU solution to invoke JNI, will be turned on by switching the config "spark.rapids.sql.fast.sample". 
	This solution is inconsistent with CPU exec although it's faster.
   This solution is 3x faster than the previous.
	Test Deails
		Data set: sample 1% from 70,000,000 rows
		Result:
			1. CPU samples and GPU gathers(Previous solution): avg 275 ms
			2. GPU sample JNI:  used time: avg 81ms
Others
	Add the metrics for all the scenarios
	Fixed the bug: When Columnar Bach has no columns, throws an exception.

Note: depends on CUDF issue: rapidsai/cudf#9728

[res-life]

build

[res-life]

Build failed because of CUDF PR is not merged.

[revans2]

Marking as draft because the CUDF JNI dependency is not in yet.

[revans2]

This is mostly for my own curiosity. You could have also done this with

withResource(cudf.ColumnVector.fromInts(rows: _*)) { gatherCv =>
  withResource(GpuColumnVector.from(cb)) { table =>
    // GPU gather
    withResource(table.gather(gatherCv)) { gatheredTable =>
      GpuColumnVector.from(gatheredTable, colTypes)
    }
  }
}

Did you try this? If so was your current way more performant? I realize it might involve an extra memory copy because the ArrayBuffer needs to be transformed into an Array, so it can be passed to fromInts. So if you didn't I don't think it is worth spending too much time on it.

[res-life]

Updated, code is more neat and more effecient.

[revans2]

nit: I think it would be cleaner if we had a GpuFastSampleExec and a GpuSampleExec, then we can select which one to use when we replace it. Part of this is because creating a RapidsConf is not cheap.

[res-life]

Yes, done.

[revans2]

nit: Because we use the same seed every time then I think all of the batches will be sampled the same way. I am not sure mathematically how that all works out. Could we at least set the seed to seed + index so each task does it slightly differently?

[res-life]

Updated, use the way like CPU to set the seed.

[res-life]

The test result is as follow:
data set
	parquet file only has one int column, one partition, has 2,100,000,000 rows.
config is
	spark.executor.cores=1
	spark.rapids.sql.concurrentGpuTasks=1
	// limit to one thead
sql is:
	spark.read.parquet("/home/chong/data/parquet").sample(0.01, 0).agg(f.sum("salary")).show()
result
	CPU used time: 4.6s
	GPU: 5.1s  		   // 	spark.conf.set("spark.rapids.sql.fast.sample", True)
	CPU GPU hybrid: 11.41s  // 	spark.conf.set("spark.rapids.sql.fast.sample", False)
conclusion
	GPU is a little slow than CPU, maybe we need to file a CUDF issue to improve the performance of the sample JNI

[res-life]

build

[res-life]

@revans2 Please see the above test result; Please help to review.

[revans2]

I get very different results, and I think your issue is because of Parquet reading.

I ran spark.time(spark.range(Int.MaxValue).sample(0.01, 0).count)

First single task/thread. This is the median of 3 runs, and measured in ms.

CPU	GPU	FAST GPU
10633	11108	83

It is also clear from the op time metrics on the UI for the runs.

When I run with 12 tasks/threads I get

CPU	GPU	FAST GPU
1236	2952	100

The regular GPU being slower is because I have my GPU semaphore set to 4 in parallel. So it cannot fully utilize the CPU when running.

If I change the query to include replacement spark.time(spark.range(Int.MaxValue).sample(0.01, 0).count) I get.

CPU	GPU	FAST GPU
11135	22556	116

So at this point the only thing that is lacking here is some documentation to let people know that there is a crazy fast version that is not the same as the Spark version.

[res-life]

The following optimization makes counting a data frame very fast.

GpuFastSampleExec
  } else if (cb.numCols() == 0) {
	  // for count agg, num of cols is 0
	  val c = GpuColumnVector.emptyBatchFromTypes(colTypes)
	  c.setNumRows(numSampleRows.toInt)  // can skip GPU sample.
	  c
  } else {

When querying sum the CPU is a little faster.

[revans2]

Yup I got ahead of myself and did things too quickly. You are 100% correct. Could you file a follow on issue to then look at what we can do to possibly speed up sampling on the GPU?

[revans2]

To be clear the issue appears to be related to doing it without replacement. With replacement is really fast. This indicates that thrust::shuffle_copy is the reason for the slowness.

spark.time(spark.range(Int.MaxValue * 12L).sample(true, 0.01, 0).selectExpr("SUM(id)", "COUNT(id)").show())
+-------------------+---------+
|            sum(id)|count(id)|
+-------------------+---------+
|3320410258800588221|257697960|
+-------------------+---------+

Time taken: 670 ms

[res-life]

build

[res-life]

build

[res-life]

The issue filed rapidsai/cudf#9834

[res-life]

Updated the doc, please review again @revans2