Spark 24914 working

[attilapiros]

No description provided.

[bersprockets]

It took me a little while to get my reproduction case (inappropriate broadcast that results in OOM) working with Spark 3.0. So I got started a little late.

Anyway, I've started playing around and reviewing, and I will continue tomorrow.

For my parquet test case, I noticed that the deserFactor seemed rather low (it was 2, when I expected more like 10). For the orc case, it seemed spot on (751).

This is my reprod case for parquet:

// ./bin/spark-shell --driver-memory 12g
// create data
import scala.util.Random
val rand = new Random
val df = Seq.tabulate(5000000) { x => ("hello there", 20, 20, 20, rand.nextLong % 20000) }.toDF("a", "b", "c", "d", "e")
df.createOrReplaceTempView("df")
sql("drop table if exists fred")
sql("create table fred stored as parquet as select * from df")

val df = spark.range(0, 2000000).map(x => (x, x + 1, x + 2, x + 3)).toDF("a", "b", "c", "d")
df.createOrReplaceTempView("df")
sql("drop table if exists phil")
sql("create table phil stored as parquet as select * from df")


// ./bin/spark-shell --master "local-cluster[6, 1, 4096]" --driver-memory 1g
// run test
sql("select f.a, f.b, f.c, f.d, p.c from fred f, phil p where f.b = p.a").show

This is my reprod case with ORC. In this case, I had a smaller TextFile table (but whose files are bigger), and a bigger ORC table(but whose files are smaller)

// ./bin/spark-shell --driver-memory 12g
// create data
val df = spark.range(0, 5000000).map { x => (x, x + 1, x + 2, x + 3)}.toDF("a", "b", "c", "d")
df.createOrReplaceTempView("df")
sql("drop table if exists fredorc")
sql("create table fredorc stored as orc as select * from df")

val df = spark.range(0, 2000000).map { x => (x, x + 1, x + 2, x + 3)}.toDF("a", "b", "c", "d")
df.createOrReplaceTempView("df")
sql("drop table if exists philtxt")
sql("create table philtxt stored as textfile as select * from df")


// ./bin/spark-shell --master "local-cluster[6, 1, 4096]" --driver-memory 1g
// run test
sql("select f.a, f.b, f.c, f.d, p.c from fredorc f, philtxt p where f.b = p.a").show

[bersprockets]

Why did this change?

[attilapiros]

as the result is not just total size (but I can revert this)

[attilapiros]

you probably meant the indentation so I have reverted that back

[bersprockets]

I meant sparkSession.sessionState vs sessionState.

[bersprockets]

I'm not sure we want to apply deserFactor to size when CBO is enabled, as I think that code already tries to calculate the size based on expected column size and number of rows (not based on raw file size). You might want to check on how EstimationUtils.getOutputSize calculates size and then see if you really needs this here.

[attilapiros]

Thanks! Great finding! :)

[bersprockets]

We might need to touch up the description a little bit, but I can't think of anything good at this time of night :).

For this one piece, maybe "This factor is calculated for columnar storage formats as a ratio of actual data size to raw file size".

[bersprockets]

Thinking out loud:
"Spark uses this ratio to scale up the estimated size, which leads to better estimate of in-memory data size and improves the query optimization (i.e., join strategy). Spark stores a ratio, rather than the data size, so that the table can grow without having to recompute statistics"

[bersprockets]

The other thing you could do is ignore the deserFactor when calculating statistics at query time if this config setting is not true. Maybe that makes the code messy (having to check SQLConf in inconvenient places).

[attilapiros]

@bersprockets Thanks for the help!

Regarding parquet I have generated the table at my place and iIt is 8 files:

$ stat -f "%N %Z" *.parquet
part-00000-633caa1d-0199-419f-b623-1ecac7d62328-c000.snappy.parquet 4004493
part-00001-633caa1d-0199-419f-b623-1ecac7d62328-c000.snappy.parquet 4003644
part-00002-633caa1d-0199-419f-b623-1ecac7d62328-c000.snappy.parquet 4003780
part-00003-633caa1d-0199-419f-b623-1ecac7d62328-c000.snappy.parquet 4003712
part-00004-633caa1d-0199-419f-b623-1ecac7d62328-c000.snappy.parquet 4003696
part-00005-633caa1d-0199-419f-b623-1ecac7d62328-c000.snappy.parquet 4003772
part-00006-633caa1d-0199-419f-b623-1ecac7d62328-c000.snappy.parquet 4003744
part-00007-633caa1d-0199-419f-b623-1ecac7d62328-c000.snappy.parquet 4003704

Now if you take the number of rows 2000000 based on:

val df = spark.range(0, 2000000).map(x => (x, x + 1, x + 2, x + 3)).toDF("a", "b", "c", "d")

And multiply it by num columns (4) and with the mem size used to store one bigint 8 bytes then divide it by the number of files you got: 8 000 000 bytes.
Which is bit under the double of a file size, like 4 003 704.

So deserFact=2 seams to me the right value.

[bersprockets]

val df = spark.range(0, 2000000).map(x => (x, x + 1, x + 2, x + 3)).toDF("a", "b", "c", "d")

I was thinking more of this case, where there is a lot of repeating data:

import scala.util.Random
val rand = new Random
val df = Seq.tabulate(5000000) { x => ("hello there", 20, 20, 20, rand.nextLong % 20000) }.toDF("a", "b", "c", "d", "e")

In this case, the ratio is more like 10, but the deserFactor is calculated as 2.

[bersprockets]

The test implies that size of data impacts the deserFactor (ratio), when in reality (I believe) it's the make-up of the data: Whether data is incrementing sequentially, or is repeating, or is random. Is that not the case?

[bersprockets]

We could instead just check that sizeInBytes == origSizeInBytes.

[attilapiros]

I would keep this assert as it is follows the same pattern used at previous cases, too

[bersprockets]

Not sure how important this is for the tests, but my memory from an escalation is that Spark calculates sizeInBytes differently depending on whether the table is partitioned or not.

Here's my recollection (for ORC, anyway):

• Not partitioned: Spark uses totalSize
• Partitioned and spark.sql.hive.convertMetastoreOrc=true: Spark prunes the partitions and sums the size of the files in the relevant partitions.
• Partitioned and spark.sql.hive.convertMetastoreOrc=false: : Spark uses totalSize

[bersprockets]

I was trying to figure out a few things, since my memory has faded. Here are my notes as I looked at some code (in the context of parquet tables, but I assume the same applies for ORC):

Non-partitioned Hive table with statistics

Spark gets the statistics from totalSize in the Hive table.

When the user inserts new rows into the table, Spark updates totalSize in Hive (at least it does on my laptop). I assume insert operations from Hive will also update totalSize (although I have not tested that).

However, if you do something whacky like the following, the statistics get out-of-date (as you would expect):

import scala.util.Random
val rand = new Random
val df = spark.range(0, 5000000).map { x => ("hello there", 20, 20, 20, rand.nextLong % 20000) }.toDF("a", "b", "c", "d", "e")
df.write.mode("append").format("parquet").save("spark-warehouse/desertest")

Non-partitioned Hive table with no statistics

In this case, Spark gets the sizeInBytes from the HadoopFsRelation (relation.sizeInBytes) in the LogicalRelation, which in turn gets the table size from location (a FileIndex, or more specifically, InMemoryFileIndex).

Note that HadoopFsRelation.sizeInBytes will return location.sizeInBytes multiplied by the setting for spark.sql.sources.fileCompressionFactor (default 1.0). spark.sql.sources.fileCompressionFactor is similar in concept to our deserFactor, except it is across the board for all tables that have no statistics.

So it seems that when a Hive table has no statistics, Spark gets sizeInBytes from the sum of the current file sizes times the fileCompressionFactor. I guess in this case Spark will always be up-to-date with the file sizes. That is, when files are dumped into the tables directory (e.g., my example above), sizeInBytes will reflect the change.

If we had an external tool to calculate deserFactor, or an option on ANALYZE TABLE, we could provide a deserFactor in this case as well.

Note, that when spark.sql.hive.convertMetastoreParquet=false and the table has no statistics, Spark just sets sizeInBytes to a giant default size (9223372036854775807).

[attilapiros]

unfortunately the parquet getTotalUncompressedSize does not reflect the size of the data in mem: like for PLAIN_DICTIONARY encoding it is really just a counter and the value size not the multiplication of those two...

probably in the first round we will target only the ORC format

[bersprockets]

probably in the first round we will target only the ORC format

Got it.