voonhous commented on code in PR #19923:
URL: https://github.com/apache/hudi/pull/19923#discussion_r3996162264
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/HoodieSparkSqlTestBase.scala:
##########
@@ -108,33 +140,107 @@ class HoodieSparkSqlTestBase extends FunSuite with
BeforeAndAfterAll {
}
override protected def test(testName: String, testTags: Tag*)(testFun: =>
Any /* Assertion */)(implicit pos: source.Position): Unit = {
- super.test(testName, testTags: _*)(
+ super.test(testName, testTags: _*)({
+ // Held for the whole test, cleanup included, so an ExclusiveSuite never
overlaps it.
+ val readLock = HoodieSparkSqlTestBase.suiteLock.readLock()
+ readLock.lock()
try {
- testFun
- } finally {
- // The INMEMORY index keeps a JVM-static record-location map; reset it
after every test so
- // stale keys from an earlier test cannot misroute writes in a later
one. withRecordType
- // clears it between record-type iterations, but only on success and
only for tests that use
- // 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)
+ try {
+ if (sharedSessionEnabled) {
+ bindSuiteSession()
}
+ testFun
+ } finally {
+ // The INMEMORY index keeps a JVM-static record-location map; reset
it after every test so
+ // stale keys from an earlier test cannot misroute writes in a later
one. withRecordType
+ // clears it between record-type iterations, but only on success and
only for tests that use
+ // 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. In shared mode this is
+ // a no-op, see clearInMemoryIndex.
+ clearInMemoryIndex()
+ dropSuiteTables()
}
+ } finally {
+ readLock.unlock()
}
- )
+ })
+ }
+
+ /**
+ * Per-suite mode: reset the JVM-static INMEMORY index between tests (see
the note in test()).
+ * Shared mode: the index is keyed per table and dropSuiteTables clears each
dropped table's
+ * entry, so a global clear would only wipe other suites' tables.
+ */
+ protected def clearInMemoryIndex(): Unit = {
Review Comment:
Right, containsOrBeforeTimelineStarts accepts an instant older than the
current timeline, so the guard cannot cover a path reused by a re-created
table; the PR text leaned on it and should not have. 7145d037ffa7 says so in
the index javadoc and in clearInMemoryIndex's scaladoc (a test that re-creates
a table at the same path must clear that path itself), and the PR description
is corrected. Latent today, as you found: every INMEMORY leg uses a fresh name
and directory.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/ExclusiveSuite.scala:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.common
+
+import org.scalatest.{Args, Status}
+
+/**
+ * Marks a suite that must not run at the same time as any other suite in the
JVM because it
+ * mutates JVM-wide state (the shared Hadoop conf, persisted RDDs of the
shared context, a
+ * static metrics registry). Takes the write side of
[[HoodieSparkSqlTestBase.suiteLock]] for
+ * its whole run; every other suite holds the read side while a test runs. No
effect until
+ * suites run concurrently.
+ */
+trait ExclusiveSuite extends HoodieSparkSqlTestBase {
+
+ abstract override def run(testName: Option[String], args: Args): Status = {
Review Comment:
Not intentional, thanks. 7145d037ffa7 moves the read side to the suite's
whole run (an override of run in the base), so beforeAll, the lazy session init
with its Hadoop-conf mutation, and afterAll are covered, and the per-test lock
is gone. The cost is that an exclusive suite waits for the suites already
running to finish rather than for their current test, bounded by the longest
suite and paid once per shard; the reshard step later moves exclusive suites to
the serial job anyway.
--
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]
