voonhous commented on code in PR #19921:
URL: https://github.com/apache/hudi/pull/19921#discussion_r3995937811
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/HoodieSparkSqlTestBase.scala:
##########
@@ -71,25 +75,50 @@ class HoodieSparkSqlTestBase extends FunSuite with
BeforeAndAfterAll {
// is consistent with the fixtures
DateTimeZone.setDefault(DateTimeZone.UTC)
TimeZone.setDefault(DateTimeUtils.getTimeZone("UTC"))
- protected lazy val spark: SparkSession = SparkSession.builder()
- .config("spark.sql.warehouse.dir", sparkWareHouse.getCanonicalPath)
- .config("spark.sql.session.timeZone", "UTC")
- .config("hoodie.insert.shuffle.parallelism", "4")
- .config("hoodie.upsert.shuffle.parallelism", "4")
- .config("hoodie.delete.shuffle.parallelism", "4")
- .config(sparkConf())
- .getOrCreate()
+ protected lazy val spark: SparkSession = if (sharedSessionEnabled) {
+ val session = HoodieSparkSqlTestBase.sharedBaseSession().newSession()
+ SparkSession.setActiveSession(session)
+ applySuiteConfToSharedSession(session)
+ session
+ } else {
+ HoodieSparkSqlTestBase.sessionBuilder(sparkWareHouse,
sparkConf()).getOrCreate()
+ }
private var tableId = new AtomicInteger(0)
private var extraConf = Map[String, String]()
+ // Shared mode: spark.hadoop.* keys this suite set on the shared Hadoop
conf, with the value they replaced.
+ private var hadoopConfOverrides: Seq[(String, String)] = Seq.empty
+
def sparkConf(): SparkConf = {
val conf = getSparkConfForTest("Hoodie SQL Test")
conf.setAll(extraConf)
conf
}
+ /**
+ * Shared mode: the context-level SparkConf is fixed, so the deltas a suite
adds through extraConf or a
+ * sparkConf() override go to its session conf (hoodie.* and spark.sql.*
keys), except spark.hadoop.*
+ * keys, which the write client reads from sparkContext.hadoopConfiguration
and which are restored in
+ * afterAll. Any other spark.* key is a SparkContext setting that a child
session cannot change, so it
+ * is rejected here rather than accepted into the session conf with no
effect.
+ */
+ private def applySuiteConfToSharedSession(session: SparkSession): Unit = {
+ val defaults = getSparkConfForTest("Hoodie SQL Test").getAll.toMap
+ val hadoopConf = session.sparkContext.hadoopConfiguration
+ sparkConf().getAll.filterNot { case (k, v) => defaults.get(k).contains(v)
}.foreach {
+ case (k, v) if k.startsWith("spark.hadoop.") =>
Review Comment:
Right: getAll gives no order, so a suite carrying a spark.hadoop.* key next
to a rejected one could mutate the shared Hadoop conf first, and afterAll would
then re-enter the failed lazy val and record the mutated value as the previous
one. 1b717265facb partitions the deltas and rejects (now including static SQL
confs via SQLConf.isStaticConfigKey) before anything is applied, and afterAll
restores through the base session instead of the lazy val.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/HoodieSparkSqlTestBase.scala:
##########
@@ -118,23 +150,68 @@ class HoodieSparkSqlTestBase extends FunSuite with
BeforeAndAfterAll {
// it, so a throwing or non-withRecordType INMEMORY test would
otherwise leak state here.
// Runs before the catalog cleanup so it holds even if a drop throws.
HoodieInMemoryHashIndex.clear()
- val catalog = spark.sessionState.catalog
- catalog.listDatabases().foreach { db =>
- catalog.listTables(db).foreach { table =>
- catalog.dropTable(table, true, true)
- }
- }
+ dropSuiteTables()
}
)
}
+ /**
+ * Shared mode: scalatest runs each suite on its own thread and Hudi reads
SparkSession.active in a few
+ * places (HoodieCatalog captures it when the session's catalog is first
built, BaseProcedure per CALL),
+ * so pin this suite's child session to the test thread and fail fast if the
session's HoodieCatalog was
+ * built against another session.
+ */
+ private def bindSuiteSession(): Unit = {
+ SparkSession.setActiveSession(spark)
+
spark.sessionState.catalogManager.catalog(CatalogManager.SESSION_CATALOG_NAME)
match {
+ case hoodieCatalog: HoodieCatalog =>
+ assert(hoodieCatalog.spark eq spark,
+ s"HoodieCatalog of ${getClass.getSimpleName} is bound to another
SparkSession")
+ case _ =>
+ }
+ }
+
+ private lazy val tableNamePrefix: String =
s"h${getClass.getSimpleName.toLowerCase}_"
+
+ /**
+ * Drops the tables a test left behind. Per-suite mode owns the whole
catalog. Shared mode shares the
+ * external catalog with every other suite in the JVM, so it drops only this
suite's generateTableName
+ * tables plus any table whose name is not of that form (fixed names and
temp views); under serial
+ * execution those can only come from the test that just ran. Fixed names
are renamed in a later step.
+ */
+ private def dropSuiteTables(): Unit = {
+ val catalog = spark.sessionState.catalog
+ catalog.listDatabases().foreach { db =>
+ val tables = catalog.listTables(db)
+ val toDrop = if (sharedSessionEnabled) tables.filter(table =>
ownsTable(table.table)) else tables
+ toDrop.foreach(table => catalog.dropTable(table, true, true))
+ }
+ }
+
+ /** Shared mode: a table is this suite's if generateTableName produced it,
or if no suite's generateTableName could have. */
+ private def ownsTable(name: String): Boolean = {
+ name.startsWith(tableNamePrefix) ||
!HoodieSparkSqlTestBase.GENERATED_TABLE_NAME.matcher(name).matches()
+ }
+
protected def generateTableName: String = {
s"h${getClass.getSimpleName.toLowerCase}_${tableId.incrementAndGet()}"
Review Comment:
Done in 1b717265facb: generateTableName builds on tableNamePrefix, which now
sits right above it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
