This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 2688d47954e9 [SPARK-58826][SQL] Hoist repeated session.implicits
imports in AutoCDC tests
2688d47954e9 is described below
commit 2688d47954e9ba82cc832ec02d9578c2e16acdfc
Author: Szehon Ho <[email protected]>
AuthorDate: Tue Aug 18 16:35:52 2026 -0700
[SPARK-58826][SQL] Hoist repeated session.implicits imports in AutoCDC tests
### What changes were proposed in this pull request?
Hoist repeated per-test `val session = spark; import session.implicits._`
blocks to a single class-level `import testImplicits._` in the remaining
AutoCDC test suites that still used the per-test pattern.
Suites updated (12 files):
- `AutoCdcFlowSuite`
- `AutoCdcOutOfOrderConvergenceSuite`
- `AutoCdcScd1AuxiliaryTableDurabilitySuite`
- `AutoCdcScd1AuxiliaryTableSpecSuite`
- `AutoCdcScd1FullRefreshSuite`
- `AutoCdcScd1MultiPipelineSuite`
- `AutoCdcScd1SchemaEvolutionSuite`
- `AutoCdcScd1SinglePipelineSuite`
- `AutoCdcScd1TargetTableDurabilitySuite`
- `AutoCdcScd2AuxiliaryTableSpecSuite`
- `AutoCdcScd2ColumnEvolutionSuite`
- `AutoCdcScd2SinglePipelineSuite`
### Why are the changes needed?
Several AutoCDC suites already use a class-level `import testImplicits._`.
The remaining suites still repeated the `session.implicits` import in every
test (or helper), which adds noise without changing behavior.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
```
SPARK_LOCAL_IP=127.0.0.1 build/sbt 'pipelines/testOnly *AutoCdc*'
```
All 193 AutoCDC tests passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.5
Closes #58052 from szehon-ho/cleanup-autocdc-testimplicits.
Authored-by: Szehon Ho <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit 6f67160ff1787bf9c922f27060025e9bca75a0df)
Signed-off-by: Szehon Ho <[email protected]>
---
.../sql/pipelines/autocdc/AutoCdcFlowSuite.scala | 10 +----
.../graph/AutoCdcOutOfOrderConvergenceSuite.scala | 5 +--
.../AutoCdcScd1AuxiliaryTableDurabilitySuite.scala | 23 +----------
.../graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala | 4 +-
.../graph/AutoCdcScd1FullRefreshSuite.scala | 11 +-----
.../graph/AutoCdcScd1MultiPipelineSuite.scala | 17 +--------
.../graph/AutoCdcScd1SchemaEvolutionSuite.scala | 44 +---------------------
.../graph/AutoCdcScd1SinglePipelineSuite.scala | 14 +------
.../AutoCdcScd1TargetTableDurabilitySuite.scala | 11 +-----
.../graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala | 4 +-
.../graph/AutoCdcScd2ColumnEvolutionSuite.scala | 17 +--------
.../graph/AutoCdcScd2SinglePipelineSuite.scala | 10 +----
12 files changed, 24 insertions(+), 146 deletions(-)
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
index 4596e31d778a..afe1915a22ab 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
@@ -44,6 +44,8 @@ import org.apache.spark.sql.types.{DataType, IntegerType,
LongType, StringType,
*/
class AutoCdcFlowSuite extends QueryTest with SharedSparkSession {
+ import testImplicits._
+
private val testIdentifier = TableIdentifier("cdc_target", Some("db"))
/** A no-op [[FlowFunction]] that throws if invoked; AutoCdcFlow tests
should never call it. */
@@ -184,8 +186,6 @@ class AutoCdcFlowSuite extends QueryTest with
SharedSparkSession {
/** A stable 3-column source streaming dataframe used across most schema
tests. */
private def threeColumnSourceDf(): DataFrame = {
- val session = spark
- import session.implicits._
MemoryStream[(Int, String, Option[Long])].toDS().toDF("id", "name", "seq")
}
@@ -416,8 +416,6 @@ class AutoCdcFlowSuite extends QueryTest with
SharedSparkSession {
test("AutoCdcMergeFlow.schema's SCD2 framework columns use the resolved
sequencing type") {
// A non-Long sequencing expression must flow through to __START_AT /
__END_AT and the
// metadata struct's record-start-at field.
- val session = spark
- import session.implicits._
val sourceDf = MemoryStream[(Int, String, Int)].toDS().toDF("id", "name",
"seq")
val resolvedFlow = newAutoCdcMergeFlow(sourceDf, storedAsScdType =
ScdType.Type2)
@@ -434,8 +432,6 @@ class AutoCdcFlowSuite extends QueryTest with
SharedSparkSession {
// A multi-column sequence is expressed as a struct over the ordering
columns; its resolved
// type is the corresponding StructType, which must flow into __START_AT /
__END_AT and the
// metadata struct's record-start-at field unchanged, including each
field's own nullability.
- val session = spark
- import session.implicits._
// seq1 is a non-null Int; seq2 is a nullable Long (Option[Long]), so the
struct carries
// mixed per-field nullability.
val sourceDf =
@@ -550,8 +546,6 @@ class AutoCdcFlowSuite extends QueryTest with
SharedSparkSession {
/** Builds an empty source df with `id` + `seq` + the supplied extra
columns. */
private def sourceDfWithExtraColumns(extraColumns: (String, DataType)*):
DataFrame = {
- val session = spark
- import session.implicits._
val baseStream = MemoryStream[(Int, Option[Long])].toDS().toDF("id", "seq")
extraColumns.foldLeft(baseStream) { case (acc, (name, dt)) =>
acc.withColumn(name, F.lit(null).cast(dt))
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala
index f164b2e70f8a..ef4ba62052fb 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala
@@ -58,6 +58,8 @@ class AutoCdcOutOfOrderConvergenceSuite
with SharedSparkSession
with AutoCdcGraphExecutionTestMixin {
+ import testImplicits._
+
// Distinct keys in the generated event stream.
private val numDistinctKeys: Int = 5
// Upper bound on unique events (one per sequence) generated per key, before
intentionally
@@ -177,9 +179,6 @@ class AutoCdcOutOfOrderConvergenceSuite
}
private def runConvergenceTest(seed: Long, scdType: ScdType): Unit = {
- val session = spark
- import session.implicits._
-
val rand = new Random(seed)
val sortedEventStream = generateRandomCdcEventStream(rand)
val shuffledEventStream = rand.shuffle(sortedEventStream)
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
index 5d93de6f9efe..73ef85c50189 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
@@ -39,10 +39,9 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
with SharedSparkSession
with AutoCdcGraphExecutionTestMixin {
- test("a higher-sequence event in a later pipeline run correctly upserts the
row") {
- val session = spark
- import session.implicits._
+ import testImplicits._
+ test("a higher-sequence event in a later pipeline run correctly upserts the
row") {
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$scd1MetadataDdl)"
@@ -82,9 +81,6 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
test("an event with a sequence lower than what was applied in a prior
pipeline run " +
"is suppressed") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$scd1MetadataDdl)"
@@ -120,9 +116,6 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
test("the auxiliary table places the AutoCDC key column first, ahead of any
non-key " +
"source columns") {
- val session = spark
- import session.implicits._
-
// Source DF column order is (name, id, version): the AutoCDC key column
`id` does NOT
// appear first in the source DF. The auxiliary table must still write
`id` as its
// leading column.
@@ -150,9 +143,6 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
test("the auxiliary table preserves the user's declared key order,
independent of the " +
"source DataFrame and target table column orders") {
- val session = spark
- import session.implicits._
-
// Source DF: (value, id, region, version). Target table: (value, id,
region, version,
// _cdc_metadata) -- same ordering as the source. The user, however,
declares
// `keys = Seq("region", "id")` -- the OPPOSITE order from how those
columns appear in
@@ -183,9 +173,6 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
test("a dry run resolves and validates the graph without provisioning the
auxiliary " +
"table") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
@@ -208,9 +195,6 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
test("if the AutoCDC auxiliary table is dropped between runs, it is
transparently " +
"recreated") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
@@ -247,9 +231,6 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
test("auxiliary key-column-names property survives identifiers containing
special " +
"characters that exercise both JSON and SQL string-literal escaping") {
- val session = spark
- import session.implicits._
-
// This test exercises the full identifier-text persistence path with
composite keys whose
// names collectively cover every escape class:
// - `it's` -- single quote: not escaped by JSON; the
writer must double it
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
index 6de8cf52f7b4..8bd2f2a0d1e8 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
@@ -43,12 +43,12 @@ import org.apache.spark.sql.types.LongType
*/
class AutoCdcScd1AuxiliaryTableSpecSuite extends PipelineTest with
SharedSparkSession {
+ import testImplicits._
+
private def targetIdentifier = fullyQualifiedIdentifier("target")
/** Source change feed with data columns `(id, name, version)`. */
private def sourceDf = {
- val session = spark
- import session.implicits._
val stream = MemoryStream[(Int, String, Long)]
stream.addData((1, "alice", 1L))
stream.toDF().toDF("id", "name", "version")
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1FullRefreshSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1FullRefreshSuite.scala
index 549c79116d36..d7c676a21117 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1FullRefreshSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1FullRefreshSuite.scala
@@ -37,10 +37,9 @@ class AutoCdcScd1FullRefreshSuite
with SharedSparkSession
with AutoCdcGraphExecutionTestMixin {
- test("full refresh wipes target rows and the auxiliary table for the
refreshed flow") {
- val session = spark
- import session.implicits._
+ import testImplicits._
+ test("full refresh wipes target rows and the auxiliary table for the
refreshed flow") {
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$scd1MetadataDdl)"
@@ -96,9 +95,6 @@ class AutoCdcScd1FullRefreshSuite
test("after a full refresh, an event with a sequence below the previous
run's " +
"watermark now lands") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$scd1MetadataDdl)"
@@ -157,9 +153,6 @@ class AutoCdcScd1FullRefreshSuite
}
test("selective full refresh wipes only the requested target's auxiliary
state") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.t_a " +
s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
index 7191f2e60cd0..c24f7283af2a 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
@@ -38,11 +38,10 @@ class AutoCdcScd1MultiPipelineSuite
with SharedSparkSession
with AutoCdcGraphExecutionTestMixin {
+ import testImplicits._
+
test("two AutoCDC pipelines targeting separate tables maintain independent
target and " +
"auxiliary tables") {
- val session = spark
- import session.implicits._
-
// Two distinct target tables created up-front.
spark.sql(
s"CREATE TABLE $catalog.$namespace.t_a " +
@@ -92,9 +91,6 @@ class AutoCdcScd1MultiPipelineSuite
test("a downstream pipeline can read an AutoCDC target written by a
different pipeline " +
"without observing the CDC metadata column") {
- val session = spark
- import session.implicits._
-
// Pipeline #1 writes into target `src` via AutoCDC.
spark.sql(
s"CREATE TABLE $catalog.$namespace.src " +
@@ -130,9 +126,6 @@ class AutoCdcScd1MultiPipelineSuite
test("two AutoCDC pipelines targeting the same table with identical key and
data " +
"schemas merge into a shared target table") {
- val session = spark
- import session.implicits._
-
// Target table is created once up-front; both pipelines target it with
the same
// AutoCDC `keys` and the same source-DF data schema. The two pipelines
have distinct
// flow names ("flow_v1" / "flow_v2") so they own independent streaming
checkpoints,
@@ -189,9 +182,6 @@ class AutoCdcScd1MultiPipelineSuite
test("two AutoCDC pipelines targeting the same table with the same key but
different " +
"data columns evolve the shared target schema") {
- val session = spark
- import session.implicits._
-
// Target is created up-front with pipeline #1's schema only; pipeline #2
brings a new
// top-level nullable `age` column that the dataset materialization layer
is expected
// to schema-merge into the target.
@@ -260,9 +250,6 @@ class AutoCdcScd1MultiPipelineSuite
test("a second pipeline targeting an existing AutoCDC table with different
keys " +
"fails with KEY_SCHEMA_DRIFT") {
- val session = spark
- import session.implicits._
-
// Target table with both candidate keys present so the second pipeline
would otherwise
// be schema-compatible with the first; only the AutoCDC `keys` differ
between flows.
spark.sql(
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
index 1a626a6ddcdf..448820443d13 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
@@ -48,10 +48,9 @@ class AutoCdcScd1SchemaEvolutionSuite
with SharedSparkSession
with AutoCdcGraphExecutionTestMixin {
- test("a nullable non-key column merges correctly with mixed NULL and
non-NULL values") {
- val session = spark
- import session.implicits._
+ import testImplicits._
+ test("a nullable non-key column merges correctly with mixed NULL and
non-NULL values") {
// Single MemoryStream with `email` as nullable from the start. Run #1
emits a row with
// a NULL email; run #2 emits an upsert with a non-NULL email.
spark.sql(
@@ -87,9 +86,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("widening a non-key column's type between runs fails with " +
"CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE") {
- val session = spark
- import session.implicits._
-
// Changing a non-key column's type between pipeline runs is rejected by
// `SchemaMergingUtils` with CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE even when
the new type
// is strictly wider. Users must full-refresh the target to change column
types.
@@ -131,9 +127,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("narrowing a non-key column's type between runs fails with " +
"CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE") {
- val session = spark
- import session.implicits._
-
// Mirror image of the widening test above: changing a non-key column's
type between
// pipeline runs is rejected by SchemaMergingUtils with
CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE
// even when the new type is strictly narrower.
@@ -176,9 +169,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("a new top-level nullable column appearing in the source DF between
runs is " +
"added to the target") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$scd1MetadataDdl)"
@@ -224,9 +214,6 @@ class AutoCdcScd1SchemaEvolutionSuite
}
test("additive target-column evolution leaves the SCD1 auxiliary table
schema unchanged") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
@@ -275,9 +262,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("broadening the column selection between runs adds the newly-included
column to " +
"the target") {
- val session = spark
- import session.implicits._
-
// Source DF schema is fixed at (id, name, email, version) across both
runs. Only the
// `columnSelection` knob differs: run #1 includes (id, name, version);
run #2 selects
// None (= all source columns). mergeSchemas adds `email` to the target
via the same
@@ -323,9 +307,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("narrowing the column selection between runs preserves the dropped
column on " +
"existing rows and leaves it NULL on new rows") {
- val session = spark
- import session.implicits._
-
// Validates the additive-only column-selection contract on the narrowing
side:
// tightening `columnSelection` between runs leaves the dropped column in
place at the
// schema level (SDP's `SchemaMergingUtils.mergeSchemas` is a union, never
a subtraction).
@@ -370,9 +351,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("a top-level column dropped from the source DF between runs is
preserved on " +
"existing rows and left NULL on new rows") {
- val session = spark
- import session.implicits._
-
// Symmetric to the new-source-column case (which exercises the source DF
*gaining* a
// column). Validates that the additive-only column-selection contract
holds when the
// narrowing is driven by the source DF's own schema shrinking, rather
than by a
@@ -421,9 +399,6 @@ class AutoCdcScd1SchemaEvolutionSuite
}
test("dropping a nested struct field between runs fails with
INCOMPATIBLE_DATA_FOR_TABLE") {
- val session = spark
- import session.implicits._
-
// The v2 writer's column-resolution layer requires every nested target
field to be
// present in the microbatch DF. When run #2's source projection drops
`b.c`, the merge
// fails with INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA. Users who want
to drop a
@@ -480,9 +455,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("a new field added inside an array<struct> element between runs is
added to the " +
"target") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(key INT NOT NULL, version BIGINT NOT NULL, " +
@@ -534,9 +506,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("dropping a field inside an array<struct> element between runs fails
with " +
"INCOMPATIBLE_DATA_FOR_TABLE") {
- val session = spark
- import session.implicits._
-
// Symmetric to the nested-struct case, but for `array<struct>`. The v2
writer rejects
// the merge because it cannot find data for the target's
`vals.element.b.d` column
// when run #2's projection drops `d` from the element struct. Users must
full-refresh
@@ -588,9 +557,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("a source DF column whose name differs from the target only by case is
folded onto the " +
"existing column under case-insensitive resolution") {
- val session = spark
- import session.implicits._
-
// Under case-insensitive resolution (Spark's default), a target `value`
and a source `Value`
// are the same column. Schema evolution honors that by threading
case-sensitivity into
// `SchemaMergingUtils.mergeSchemas`: the merge maps `Value` onto the
existing `value`, so
@@ -633,9 +599,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("extra columns on the target that the AutoCDC flow does not emit are
preserved " +
"across the merge") {
- val session = spark
- import session.implicits._
-
// The target is wider than the AutoCDC flow's source DF: column `extra`
is present on
// the target but never produced by the flow. AutoCDC must tolerate the
extra target
// column -- pre-existing rows keep their `extra` value, and
newly-inserted rows
@@ -670,9 +633,6 @@ class AutoCdcScd1SchemaEvolutionSuite
test("changing a non-key column type from TIMESTAMP to STRING between runs
fails with " +
"CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE") {
- val session = spark
- import session.implicits._
-
// `mergeSchemas` rejects an incompatible type change between TIMESTAMP
and STRING.
// Captured alongside the type-widening / type-narrowing tests; users must
full-refresh
// the target to change a column's type.
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SinglePipelineSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SinglePipelineSuite.scala
index 9ea7a41d0372..f37b86497505 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SinglePipelineSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SinglePipelineSuite.scala
@@ -40,10 +40,9 @@ class AutoCdcScd1SinglePipelineSuite
with SharedSparkSession
with AutoCdcGraphExecutionTestMixin {
- test("an upsert event lands a new row in an empty target table") {
- val session = spark
- import session.implicits._
+ import testImplicits._
+ test("an upsert event lands a new row in an empty target table") {
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$scd1MetadataDdl)"
@@ -73,9 +72,6 @@ class AutoCdcScd1SinglePipelineSuite
test("consecutive upsert, delete, and re-upsert events for the same key in
one run " +
"converge to the latest event") {
- val session = spark
- import session.implicits._
-
// Target schema deliberately omits `is_delete`: the source carries it as
a control
// column, drives the deleteCondition, and is excluded from the target
projection.
spark.sql(
@@ -117,9 +113,6 @@ class AutoCdcScd1SinglePipelineSuite
test("two AutoCDC flows targeting separate tables in one pipeline produce
independent " +
"results") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.t_a " +
s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
@@ -168,9 +161,6 @@ class AutoCdcScd1SinglePipelineSuite
test("an AutoCDC flow targeting a table whose format does not support
row-level " +
"operations fails with AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE") {
- val session = spark
- import session.implicits._
-
// Intentionally use a non-merge-compatible catalog, whose default table
format is parquet.
val catalog = TestGraphRegistrationContext.DEFAULT_CATALOG
val database = TestGraphRegistrationContext.DEFAULT_DATABASE
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1TargetTableDurabilitySuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1TargetTableDurabilitySuite.scala
index 0f777708398b..e3ac46db1891 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1TargetTableDurabilitySuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1TargetTableDurabilitySuite.scala
@@ -35,11 +35,10 @@ class AutoCdcScd1TargetTableDurabilitySuite
with SharedSparkSession
with AutoCdcGraphExecutionTestMixin {
+ import testImplicits._
+
test("pre-loaded rows: an event with a lower sequence is suppressed and a
higher one " +
"wins") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$scd1MetadataDdl)"
@@ -75,9 +74,6 @@ class AutoCdcScd1TargetTableDurabilitySuite
test("pre-loaded target rows merge correctly on the first AutoCDC run, and
the " +
"auxiliary table is created lazily") {
- val session = spark
- import session.implicits._
-
// Target was populated by some external process; this is the first
AutoCDC run.
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
@@ -118,9 +114,6 @@ class AutoCdcScd1TargetTableDurabilitySuite
test("a target table created without the CDC metadata column gets the column
" +
"auto-added on the first AutoCDC run") {
- val session = spark
- import session.implicits._
-
// User creates the target without the AutoCDC metadata column.
DatasetManager evolves
// the existing table schema by merging it with the AutoCdcMergeFlow's
output schema,
// which includes the metadata column. The first run therefore proceeds
normally, and
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala
index d01ec0f54736..46fbea24c6ae 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala
@@ -44,12 +44,12 @@ import org.apache.spark.sql.types.{LongType, StructField,
StructType}
*/
class AutoCdcScd2AuxiliaryTableSpecSuite extends PipelineTest with
SharedSparkSession {
+ import testImplicits._
+
private def targetIdentifier = fullyQualifiedIdentifier("target")
/** Source change feed with data columns `(id, name, version)`. */
private def sourceDf = {
- val session = spark
- import session.implicits._
val stream = MemoryStream[(Int, String, Long)]
stream.addData((1, "alice", 1L))
stream.toDF().toDF("id", "name", "version")
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2ColumnEvolutionSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2ColumnEvolutionSuite.scala
index 4daa0d5b0dd9..25cb185a7ff4 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2ColumnEvolutionSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2ColumnEvolutionSuite.scala
@@ -57,6 +57,8 @@ class AutoCdcScd2ColumnEvolutionSuite
with SharedSparkSession
with AutoCdcGraphExecutionTestMixin {
+ import testImplicits._
+
/** The SCD2 target's `_cdc_metadata` struct value for a given
recordStartAt. */
private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
@@ -66,9 +68,6 @@ class AutoCdcScd2ColumnEvolutionSuite
test("a source column dropped between runs is preserved on existing records
and null on new " +
"ones") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, name STRING, email STRING, version BIGINT NOT NULL,
$scd2MetadataDdl)"
@@ -114,9 +113,6 @@ class AutoCdcScd2ColumnEvolutionSuite
test("narrowing the COLUMNS selection to drop a non-tracked column preserves
it on existing " +
"records and leaves it null on new ones") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, name STRING, email STRING, version BIGINT NOT NULL,
$scd2MetadataDdl)"
@@ -165,9 +161,6 @@ class AutoCdcScd2ColumnEvolutionSuite
}
test("a late narrower event weaves into history without rewriting existing
records") {
- val session = spark
- import session.implicits._
-
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, name STRING, email STRING, version BIGINT NOT NULL,
$scd2MetadataDdl)"
@@ -215,9 +208,6 @@ class AutoCdcScd2ColumnEvolutionSuite
test("a nested struct field dropped between runs is preserved on existing
records and null on " +
"new ones (SCD2 is more permissive than SCD1 here)") {
- val session = spark
- import session.implicits._
-
// Contrast with SCD1: AutoCdcScd1SchemaEvolutionSuite rejects this exact
shape with
// INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA, because its MERGE source
is missing `value.b.c`
// and the v2 writer's resolver cannot find data for the target's nested
field. SCD2's
@@ -278,9 +268,6 @@ class AutoCdcScd2ColumnEvolutionSuite
test("a field dropped inside an array<struct> element between runs is
preserved on existing " +
"records and null on new ones (SCD2 is more permissive than SCD1 here)") {
- val session = spark
- import session.implicits._
-
// The array<struct> analog of the nested-struct-drop test above, and the
counterpart to
// AutoCdcScd1SchemaEvolutionSuite's array<struct> case, which fails with
// INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA on `vals.element.b.d`.
SCD2's
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala
index e413399c4033..595a3df45e28 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala
@@ -45,6 +45,8 @@ class AutoCdcScd2SinglePipelineSuite
with SharedSparkSession
with AutoCdcGraphExecutionTestMixin {
+ import testImplicits._
+
/** The SCD2 target's `_cdc_metadata` struct value for a given
recordStartAt. */
private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
@@ -62,8 +64,6 @@ class AutoCdcScd2SinglePipelineSuite
}
test("SCD2: an upsert lands an open current record in an empty target
table") {
- val session = spark
- import session.implicits._
createScd2Target(s"$catalog.$namespace.target")
val stream = MemoryStream[(Int, String, Long)]
@@ -91,8 +91,6 @@ class AutoCdcScd2SinglePipelineSuite
}
test("SCD2: an update to a key closes the prior record and opens a new one")
{
- val session = spark
- import session.implicits._
createScd2Target(s"$catalog.$namespace.target")
val stream = MemoryStream[(Int, String, Long)]
@@ -123,8 +121,6 @@ class AutoCdcScd2SinglePipelineSuite
}
test("SCD2: a delete closes the current record with no open record
remaining") {
- val session = spark
- import session.implicits._
// Target omits `is_delete`: the source carries it as a control column
driving the delete
// condition, and it is excluded from the target projection.
createScd2Target(s"$catalog.$namespace.target")
@@ -158,8 +154,6 @@ class AutoCdcScd2SinglePipelineSuite
}
test("SCD2: the auxiliary table is materialized for the target") {
- val session = spark
- import session.implicits._
createScd2Target(s"$catalog.$namespace.target")
val stream = MemoryStream[(Int, String, Long)]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
