[SPARK-29874][SQL]Optimize Dataset.isEmpty()

[AngersZhuuuu]

What changes were proposed in this pull request?

In origin way to judge if a DataSet is empty by

 def isEmpty: Boolean = withAction("isEmpty", limit(1).groupBy().count().queryExecution) { plan =>
    plan.executeCollect().head.getLong(0) == 0
  }

will add two shuffles by limit(), groupby() and count(), then collect all data to driver.
In this way we can avoid oom when collect data to driver. But it will trigger all partitions calculated and add more shuffle process.

We change it to

  def isEmpty: Boolean = withAction("isEmpty", select().queryExecution) { plan =>
    plan.executeTake(1).isEmpty
  }

After these pr, we will add a column pruning to origin LogicalPlan and use executeTake() API.
then we won't add more shuffle process and just compute only one partition's data in last stage.
In this way we can reduce cost when we call DataSet.isEmpty() and won't bring memory issue to driver side.

Why are the changes needed?

Optimize Dataset.isEmpty()

Does this PR introduce any user-facing change?

No

How was this patch tested?

Origin UT

[AngersZhuuuu]

@cloud-fan
As we discussed #26437 (comment), in this way is better？

[cloud-fan]

shall we do column pruning?

[AngersZhuuuu]

shall we do column pruning?

Of course, add it.

[cloud-fan]

Can we have some benchmark numbers?

[cloud-fan]

ok to test

[SparkQA]

Test build #113700 has finished for PR 26500 at commit 56d2093.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #113711 has finished for PR 26500 at commit 908a39f.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[maropu]

Can you address more in the PR desc for better commit logs, how to optimize it?

[srowen]

Ping @AngersZhuuuu

[AngersZhuuuu]

Ping @AngersZhuuuu

Thank you ping, sorry for pending this work. A little busy these days.
Starting work on these things.

[AngersZhuuuu]

  test("benchmark of empty") {
    var start = System.currentTimeMillis()
    var isEmpty = spark.range(10000000)
      .repartition(100)
      .limit(1)
      .groupBy()
      .count()
      .queryExecution.executedPlan.executeCollect().head.getLong(0) == 0
    println(isEmpty)
    var end = System.currentTimeMillis()
    // scalastyle:off
    println(s"duration = ${end - start}")

    start = System.currentTimeMillis()
    isEmpty = spark.range(10000000)
      .repartition(100)
      .select()
      .queryExecution.executedPlan.executeTake(1) == 0
    println(isEmpty)
    end = System.currentTimeMillis()
    // scalastyle:off
    println(s"duration = ${end - start}")
  }

Result
false
duration = 7248
false
duration = 1449

@cloud-fan @maropu @srowen
The test case is simple but can mimic the behavior before and after the API change.

[cloud-fan]

great! can you enrich the PR description? Optimize Dataset.isEmpty() is good in the "Why" section but we need to put more in the "What" section. e.g. we change the implementation to avoid shuffles.

[AngersZhuuuu]

great! can you enrich the PR description? Optimize Dataset.isEmpty() is good in the "Why" section but we need to put more in the "What" section. e.g. we change the implementation to avoid shuffles.

Updated , is clear now?

[cloud-fan]

will add three shuffles by limit(), groupby() and count()

have you confirmed? groupby + count is one operator called Aggregate.

[AngersZhuuuu]

will add three shuffles by limit(), groupby() and count()

have you confirmed? groupby + count is one operator called Aggregate.

Updated, count won't trigger shuffle.

[cloud-fan]

thanks, merging to master!

[HyukjinKwon]

LGTM too