hudi-agent commented on code in PR #19921:
URL: https://github.com/apache/hudi/pull/19921#discussion_r3995870471
##########
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:
🤖 `SparkConf.getAll` is unordered, so if a suite carries both a
`spark.hadoop.*` key and a rejected `spark.*` key (or a static `spark.sql.*`
one that `conf.set` throws on), the Hadoop override may already be applied when
the throw happens. Since the lazy val fails, `afterAll` re-enters the
initializer, records the already-overridden value as "previous", and throws
again, so the shared Hadoop conf stays poisoned for every later suite in the
JVM. Would it be worth validating all keys first and only then mutating the
shared conf?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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:
🤖 nit: could this reuse `tableNamePrefix` (e.g.
`s"$tableNamePrefix${tableId.incrementAndGet()}"`)? `ownsTable` /
`GENERATED_TABLE_NAME` now depend on this exact format, so keeping a single
source for the prefix makes it harder for the two to drift apart.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
