[BUG] DeltaTableBuilder.execute() sometimes throws "Could not find active SparkSession"

[moredatapls]

Bug

Describe the problem

Steps to reproduce

import io.delta.tables.DeltaTable
import org.apache.spark.sql.{Encoders, SparkSession}

import java.nio.file.{Files, Path, Paths}

val tempDir: Path = Files.createTempDirectory("scratch")
val warehouseDir: Path = Paths.get(tempDir.toString, "warehouse")
val tableDir: Path = Paths.get(tempDir.toString, "table")

val spark = SparkSession.builder()
  .appName("scratch")
  .master("local[*]")
  .config("spark.local.ip", "127.0.0.1")
  .config("spark.driver.host", "127.0.0.1")
  .config("spark.driver.bindAddress", "127.0.0.1")
  .config("spark.sql.warehouse.dir", warehouseDir.toString)
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
  .getOrCreate()

case class Table(id: Long, text: String)

val schema = Encoders.product[Table].schema

val tableBuilder = DeltaTable.create(spark)
  .tableName("Table")
  .addColumns(schema)
  .location(tableDir.toString)

// sometimes throws org.apache.spark.sql.delta.DeltaIllegalArgumentException: Could not find active SparkSession
// even though a session has been passed into the DeltaTable.create(spark) statement
tableBuilder.execute()

Observed results

We are running some integration tests of our Spark jobs against some Delta tables which are created locally, on the fly, in the integration tests. The tables are stored on disk. The code above shows a (simplified) snippet of how we setup the Delta tables on disk for testing.

While running those test we sometimes (not always) observe that calling DeltaTableBuilder.execute() for the Delta table that we create in the tests sometimes throws an exception: org.apache.spark.sql.delta.DeltaIllegalArgumentException: Could not find active SparkSession.

This happens despite us passing the SparkSession into the DeltaTable.create(spark) method.

Expected results

If a SparkSession is passed into DeltaTable.create(spark), the exception I described above should never be thrown, since an active session exists and has been passed into the builder.

Further details

Stack trace:

org.apache.spark.sql.delta.DeltaIllegalArgumentException: Could not find active SparkSession
at org.apache.spark.sql.delta.DeltaErrorsBase.activeSparkSessionNotFound(DeltaErrors.scala:2038)
at org.apache.spark.sql.delta.DeltaErrorsBase.activeSparkSessionNotFound$(DeltaErrors.scala:2037)
at org.apache.spark.sql.delta.DeltaErrors$.activeSparkSessionNotFound(DeltaErrors.scala:2293)
at io.delta.tables.DeltaTable$.$anonfun$forName$1(DeltaTable.scala:680)
at scala.Option.getOrElse(Option.scala:189)
at io.delta.tables.DeltaTable$.forName(DeltaTable.scala:680)
at io.delta.tables.DeltaTableBuilder.execute(DeltaTableBuilder.scala:369)

From the stack trace we can see that the problem is here:

delta/core/src/main/scala/io/delta/tables/DeltaTableBuilder.scala

Lines 367 to 371 in 2041c3b

	if (DeltaTableUtils.isValidPath(tableId)) {
	DeltaTable.forPath(location.get)
	} else {
	DeltaTable.forName(this.identifier)
	}

A possible solution could be to always forward the SparkSession object that is present in the DeltaTableBuilder to DeltaTable.forPath() and DeltaTable.forName() - they both have an optional parameter that we could use. Maybe it could be changed to something like this:

if (DeltaTableUtils.isValidPath(tableId)) {
  DeltaTable.forPath(spark, location.get)
} else {
  DeltaTable.forName(spark, this.identifier)
}

Environment information

• Delta Lake version: 2.1.1
• Spark version: 3.3.0
• Scala version: 2.12.14

Willingness to contribute

The Delta Lake Community encourages bug fix contributions. Would you or another member of your organization be willing to contribute a fix for this bug to the Delta Lake code base?

• Yes. I can contribute a fix for this bug independently.
• Yes. I would be willing to contribute a fix for this bug with guidance from the Delta Lake community.
• No. I cannot contribute a bug fix at this time.

[zsxwing]

Good catch. Yep we should just use the SparkSession object in DeltaTableBuilder to create DeltaTable. Feel free to submit a PR to fix that.

By the way, do you know why active SparkSession is not available in your tests?

[moredatapls]

By the way, do you know why active SparkSession is not available in your tests?

No, I don't. It's very weird, when we are running the integration tests locally on our computers the tests work just fine. However, in CI (we're using Azure DevOps), the tests fail a lot, but sometimes succeed. I feel like it could be some race condition somewhere since SparkSession.getActiveSession() returns the active session per thread? Not sure.

[moredatapls]

@zsxwing I published a draft PR here: #1476. It was easy to fix but maybe someone can provide me some guidance on the testing, is there any test I could add anywhere? It's my first PR here

[pgrandjean]

Same issue during tests. Looking forward to having this fixed!

[moredatapls]

@pgrandjean fyi Delta 2.2.0 was released which includes the fix. I upgraded the dependency and the tests now work in CI 🥳

[pgrandjean]

awesome! thank you 👍