This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 558007d76eaf test(spark): add stored procedure coverage (#19161)
558007d76eaf is described below
commit 558007d76eaf345f759a349bd4bc6d17967708d9
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Aug 14 06:01:23 2026 -0700
test(spark): add stored procedure coverage (#19161)
* test(spark): add stored procedure coverage
* Fix incorrect assertions in stored procedure coverage tests
- FilterUtils: type-match numeric literals (long/double/ceil/floor), split
out time-zone-aware timestamp functions that the util cannot evaluate.
- CopyToTempView incremental: begin instant is an exclusive lower bound, so
the '000'..last range yields 1 record, not 2.
- LogFile merge/filter: lower hoodie.compact.inline.max.delta.commits so the
explicit run_compaction actually produces the commit instant the merged
scan requires.
- ExportInstants: fix a real bug where desc=true threw
UnsupportedOperationException because Collections.reverse ran on the
read-only list from asJava over an immutable Scala collection.
* Move export_instants fix and tests to a dedicated fix PR
The export_instants desc-ordering product fix and its regression test move
to their own fix(spark) PR; this PR stays pure test coverage of the other
stored procedures.
* Restore ExportInstantsProcedure to master, keeping the export_instants
desc fix from #19172
* review(19161): restore export_instants tests from #19172 and pin known
limitations
- Restore TestExportInstantsProcedure to master so the desc-ordering,
action-filter and invalid-folder coverage added by #19172 stays intact
(the PR would otherwise silently revert it on merge).
- Correct the inverted incremental copy_to_temp_view comment (both bounds
are inclusive on completion time; the end bound excludes the last
commit) and assert the surviving record id instead of a bare count.
- Pin the Long-column-narrowed-to-Int coercion (#19632) and the broken
array predicates (#19633) in TestHoodieProcedureFilterUtils with
assertions that flip when the bugs are fixed.
- Document why the show_logfile_records merge test needs an explicit
compaction and distinct update keys (#19634).
- sync_validate action hardcode and connection NPE are tracked in #19635
(no test change needed here).
* review(19161): nit pass -- dedupe tests, tighten assertions, extract
helpers
- copy_to_temp_view: collapse the three query-type tests into one MOR
table test; assert the read-optimized price (10 vs snapshot 99) so the
RO arm is discriminated; use orderingFields instead of the deprecated
preCombineField.
- sync_validate: assert the exact output messages (dst-first branch,
catch-up count of 2, no catch-up suffix on the shared-timeline case)
instead of vacuous contains checks; drop the mandatory-arguments test
(checkArgs is asserted ~19x across the suite already). The Catach typo
mirrors the procedure output.
- repair_clustering_plan: extract the shared table+plan setup helper;
anchor the healthy validate_delete result by asserting planned files
exist; add a NOT_FOUND_IN_PLAN case pinning that a file outside the
plan is never deleted even with need_delete; make the allow_empty_plan
case destructive (dry_run false) and assert the rewritten instant
carries an empty plan.
- show_clean_plans: assert the column-reference message so the unknown-
column branch is discriminated from a parse failure; drop the unused
insert.
- show_logfile_*: extract the shared MOR create-table helper.
- filter utils: rename the terse schema parameter/locals (s -> schema).
* review(19161): fix red sync_validate asserts, tighten coverage, pin seven
product bugs
Round 1 (correctness fixes):
- sync_validate: strip the default. qualifier the procedure never prints
(suite was 0/2 red), stabilize the catch-up count via
hoodie.parquet.small.file.limit=0 (was 3 from small-file packing), add a
source-ahead test pinning the else-branch argument order, and pin the MOR
deltacommit failure (#19635)
- show_logfile_records: exact merged/unmerged counts, assert the commit
instant the merged scan needs, and pin the HoodieDeleteBlock MatchError on
a v9 log holding data+delete blocks, for both procedures (#19634)
- copy_to_temp_view: correct the incremental range comment (OPEN_CLOSED),
read the global view through global_temp from a new session (the assert
was reading the local view), cover createGlobalTempView including the
duplicate-name error
- filter utils: pin reversed-operand coercion and the Float/Short/Byte/
Decimal literal matrix (#19632), pin unknown functions silently dropping
rows while validateFilterExpression accepts them (#19638)
- repair_clustering_plan: cover backup => false (no .repair created)
- new TestProcedureParameterImpl: equals/hashCode regression for #19167
(aborts with StackOverflowError against the pre-fix code)
Round 2 (review follow-ups):
- sync_validate: the record-count modes fail fast offline, so pin both
cause-chain shapes -- complete masks the SQLException behind an NPE
(#19635 bug 2), latestPartitions surfaces it
- show_logfile procedures: assert the single-log-file glob precondition the
MatchError pin depends on; pin show_logfile_metadata record_count and
block_type (previously asserted nowhere)
- show_clean_plans: archival-forcing test pins all-null plan fields for
archived cleans and IOException from show_cleans/show_cleans_metadata
(#19639 covers all three); field lookups by name; the vacuous
showArchived test (never archived anything) is absorbed and deleted
* review(19161): pin last uncovered arms, tighten comment precision
- show_logfile_records: pin #19634 crash 4 -- a delete-only v9 log file
NPEs (asserted over the cause chain; Spark wraps the NPE as
INTERNAL_ERROR); name both schema-resolution sites in the comment
(merged reads the last globbed file, unmerged re-reads per file)
- sync_validate: pin the never-written-destination sentinel (empty commits
timeline falls back to "0" and the whole source timeline becomes
catch-up); correct the transport-setup wording (port -1 is rejected
before any connect, not during URI parsing)
- show cleans suite: pre-archival showArchived controls (all three
procedures succeed over an empty archive, so the post-archival failures
are attributable to archival); #19639 comment updated to the retitled
all-three-procedures scope
---------
Co-authored-by: voon <[email protected]>
---
.../procedure/TestCopyToTempViewProcedure.scala | 86 +++++-
.../procedure/TestHoodieLogFileProcedure.scala | 224 ++++++++++++---
.../procedure/TestHoodieProcedureFilterUtils.scala | 310 +++++++++++++++++++++
.../procedure/TestProcedureParameterImpl.scala | 63 +++++
.../TestRepairClusteringPlanProcedure.scala | 164 ++++++++---
.../hudi/procedure/TestShowCleansProcedures.scala | 236 +++++++++++-----
.../TestValidateHoodieSyncProcedure.scala | 254 +++++++++++++++++
7 files changed, 1190 insertions(+), 147 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCopyToTempViewProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCopyToTempViewProcedure.scala
index 0466681d611d..becb0cee46de 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCopyToTempViewProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCopyToTempViewProcedure.scala
@@ -170,11 +170,93 @@ class TestCopyToTempViewProcedure extends
HoodieSparkSqlTestBase {
val row2 = spark.sql(s"""call
copy_to_temp_view(table=>'$tableName',view_name=>'$viewName',global=>true,replace=>true)""").collectAsList()
assert(row2.size() == 1 && row2.get(0).get(0) == 0)
+ // 4: the global view is visible from another session through the
global_temp database
newSession = spark.newSession()
- // 4: query view in other session
- val newViewCount = spark.sql(s"""select count(1) from
$viewName""").collectAsList()
+ val newViewCount = newSession.sql(s"""select count(1) from
global_temp.$viewName""").collectAsList()
assert(newViewCount.size() == 1 && newViewCount.get(0).get(0) == 4)
+ // 5: copyToTempView with global=true and replace=false goes through
createGlobalTempView
+ val globalViewName = generateTableName
+ val row3 = spark.sql(s"""call
copy_to_temp_view(table=>'$tableName',view_name=>'$globalViewName',global=>true,replace=>false)""").collectAsList()
+ assert(row3.size() == 1 && row3.get(0).get(0) == 0)
+ val globalViewCount = spark.newSession().sql(s"""select count(1) from
global_temp.$globalViewName""").collectAsList()
+ assert(globalViewCount.size() == 1 && globalViewCount.get(0).get(0) == 4)
+
+ // 6: re-creating the same global view without replace fails
+ val globalViewExistsErrorMsg = if (HoodieSparkUtils.gteqSpark3_4) {
+ s"[TEMP_TABLE_OR_VIEW_ALREADY_EXISTS] Cannot create the temporary view
`$globalViewName` because it already exists."
+ } else {
+ s"Temporary view '$globalViewName' already exists"
+ }
+ checkExceptionContain(s"""call
copy_to_temp_view(table=>'$tableName',view_name=>'$globalViewName',global=>true,replace=>false)""")(globalViewExistsErrorMsg)
+ }
+ }
+
+ test("Test Call copy_to_temp_view Procedure with incremental, read_optimized
and as_of_instant") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ // MOR so the read-optimized view diverges from the snapshot once an
update lands in a log file.
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | location '${tmp.getCanonicalPath}/$tableName'
+ | tblproperties (
+ | type = 'mor',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ | )
+ """.stripMargin)
+
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+ spark.sql(s"insert into $tableName select 2, 'a2', 20, 2000")
+
+ val commits = spark.sql(s"select distinct _hoodie_commit_time from
$tableName order by _hoodie_commit_time")
+ .collect().map(_.getString(0))
+ assert(commits.length == 2)
+
+ // Incremental query type requires begin/end instant times.
+ val incViewName = generateTableName
+ checkExceptionContain(
+ s"call copy_to_temp_view(table => '$tableName', view_name =>
'$incViewName', query_type => 'incremental')")(
+ "begin_instance_time and end_instance_time can not be null")
+
+ // Incremental reads filter on *completion* time, start exclusive and
end inclusive
+ // (RangeType.OPEN_CLOSED). begin => '000' sits below every completion
time, so it drops
+ // nothing; end_instance_time is the last commit's *requested* time,
which is below that
+ // commit's own completion time, so the last commit is excluded and only
the first commit's
+ // record (id = 1) survives.
+ val incResult = spark.sql(
+ s"""call copy_to_temp_view(table => '$tableName', view_name =>
'$incViewName',
+ | query_type => 'incremental', begin_instance_time => '000',
end_instance_time => '${commits.last}')"""
+ .stripMargin).collectAsList()
+ assert(incResult.size() == 1 && incResult.get(0).get(0) == 0)
+ val incRows = spark.sql(s"select id from $incViewName").collectAsList()
+ assert(incRows.size() == 1 && incRows.get(0).getInt(0) == 1)
+
+ // The update goes to a log file, so the read-optimized view (base files
only) still sees the
+ // pre-update price where a snapshot view would see 99.
+ spark.sql(s"update $tableName set price = 99 where id = 1")
+ val roViewName = generateTableName
+ val roResult = spark.sql(
+ s"""call copy_to_temp_view(table => '$tableName', view_name =>
'$roViewName',
+ | query_type => 'read_optimized')""".stripMargin).collectAsList()
+ assert(roResult.size() == 1 && roResult.get(0).get(0) == 0)
+ val roRows = spark.sql(s"select price from $roViewName where id =
1").collectAsList()
+ assert(roRows.size() == 1 && roRows.get(0).getDouble(0) == 10.0)
+
+ // The snapshot at the first commit contains only the first record.
+ val asOfViewName = generateTableName
+ val asOfResult = spark.sql(
+ s"""call copy_to_temp_view(table => '$tableName', view_name =>
'$asOfViewName',
+ | as_of_instant =>
'${commits.head}')""".stripMargin).collectAsList()
+ assert(asOfResult.size() == 1 && asOfResult.get(0).get(0) == 0)
+ val asOfRows = spark.sql(s"select id from $asOfViewName").collectAsList()
+ assert(asOfRows.size() == 1 && asOfRows.get(0).getInt(0) == 1)
}
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieLogFileProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieLogFileProcedure.scala
index 67b770636a3e..95342715896b 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieLogFileProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieLogFileProcedure.scala
@@ -19,29 +19,47 @@
package org.apache.spark.sql.hudi.procedure
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.storage.StoragePath
+import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
+
class TestHoodieLogFileProcedure extends HoodieSparkProcedureTestBase {
+
+ /**
+ * The throwable and its causes, in order. Null-terminated, and capped so a
self-referential
+ * cause cannot spin forever.
+ */
+ private def causeChain(t: Throwable): Seq[Throwable] =
+ Iterator.iterate(t)(_.getCause).takeWhile(_ != null).take(16).toSeq
+
+ private def chainTypes(t: Throwable): String =
causeChain(t).map(_.getClass.getName).mkString(" <- ")
+
+ private def createMorTable(tableName: String, tablePath: String,
tableVersion: Option[Int] = None): Unit = {
+ val versionProperty = tableVersion.map(version => s",
hoodie.write.table.version = '$version'").getOrElse("")
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long,
+ | partition long
+ |) using hudi
+ | partitioned by (partition)
+ | location '$tablePath'
+ | tblproperties (
+ | type = 'mor',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'$versionProperty
+ | )
+ """.stripMargin)
+ }
+
test("Test Call show_logfile_metadata Procedure") {
withTempDir { tmp =>
val tableName = generateTableName
val tablePath = s"${tmp.getCanonicalPath}/$tableName"
- // create table
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long,
- | partition int
- |) using hudi
- | partitioned by (partition)
- | location '$tablePath'
- | tblproperties (
- | type = 'mor',
- | primaryKey = 'id',
- | orderingFields = 'ts'
- | )
- """.stripMargin)
+ createMorTable(tableName, tablePath)
// insert data to table
spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000, 1000")
spark.sql(s"insert into $tableName select 2, 'a2', 20, 1500, 1500")
@@ -57,6 +75,11 @@ class TestHoodieLogFileProcedure extends
HoodieSparkProcedureTestBase {
assertResult(1) {
result.length
}
+ // The single log block holds the one updated record, and the block type
is version dependent:
+ // v10 native logs give PARQUET_DATA_BLOCK, v9 avro logs give
AVRO_DATA_BLOCK.
+ assertResult(1)(result.head.getInt(1))
+ assert(result.head.getString(2).endsWith("DATA_BLOCK"),
+ s"unexpected block type ${result.head.getString(2)}")
}
}
@@ -64,24 +87,7 @@ class TestHoodieLogFileProcedure extends
HoodieSparkProcedureTestBase {
withTempDir { tmp =>
val tableName = generateTableName
val tablePath = s"${tmp.getCanonicalPath}/$tableName"
- // create table
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long,
- | partition long
- |) using hudi
- | partitioned by (partition)
- | location '$tablePath'
- | tblproperties (
- | type = 'mor',
- | primaryKey = 'id',
- | orderingFields = 'ts'
- | )
- """.stripMargin)
+ createMorTable(tableName, tablePath)
// insert data to table
spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000, 1000")
spark.sql(s"insert into $tableName select 2, 'a2', 20, 1500, 1500")
@@ -100,4 +106,150 @@ class TestHoodieLogFileProcedure extends
HoodieSparkProcedureTestBase {
}
}
}
+
+ test("Test Call show_logfile_records Procedure with merge and filter") {
+ // Keep automatic cleaning off so the pre-compaction log files survive for
the merged scan. The
+ // lowered compaction trigger + explicit run_compaction below are a
workaround: the merged scan
+ // resolves the latest instant via
getCommitAndReplaceTimeline.lastInstant.get, which throws on
+ // a deltacommit-only (never compacted) MOR table; see #19634.
+ withSQLConf("hoodie.clean.automatic" -> "false",
"hoodie.compact.inline.max.delta.commits" -> "1") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ createMorTable(tableName, tablePath)
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000, 1000")
+ spark.sql(s"insert into $tableName select 2, 'a2', 20, 1500, 1000")
+ // Each update touches a distinct key on purpose: a key updated in
more than one log block
+ // makes the merged scan cast the merged record to HoodieRecordPayload
and fail with a
+ // ClassCastException, so the merge-is-a-no-op shape is the only one
that works; see #19634.
+ spark.sql(s"update $tableName set name = 'b1' where id = 1")
+ spark.sql(s"update $tableName set name = 'b2' where id = 2")
+
+ // Compaction produces a commit instant on the timeline, which the
merged scan needs to
+ // determine the latest instant time; the delta log files remain on
disk. The lowered
+ // delta-commit threshold above already compacts inline, so this call
usually has nothing
+ // left to schedule and returns no rows -- it is the fallback for the
day inline compaction
+ // stops firing. Assert the precondition the merged scan actually
depends on, otherwise it
+ // dies later with an opaque NoSuchElementException; see #19634.
+ spark.sql(s"call run_compaction(op => 'run', table =>
'$tableName')").collect()
+ assert(createMetaClient(spark,
tablePath).getActiveTimeline.getCommitAndReplaceTimeline.lastInstant.isPresent,
+ "compaction must produce a commit instant for the merged scan")
+
+ val pattern = s"$tablePath/*/*.log.*"
+
+ // Merged scan returns one row per updated key.
+ val merged = spark.sql(
+ s"""call show_logfile_records(table => '$tableName',
log_file_path_pattern => '$pattern', merge => true, limit =>
10)""".stripMargin).collect()
+ assertResult(2)(merged.length)
+
+ // Unmerged scan, used as the baseline for the filter assertions.
Because every key is
+ // updated exactly once, merging is a no-op here and both scans return
the same rows, so
+ // this asserts no merge-vs-no-merge output distinction: a shape that
would show one needs a
+ // key updated in more than one log block, which the merged scan
cannot read; see #19634.
+ val unfiltered = spark.sql(
+ s"""call show_logfile_records(table => '$tableName',
log_file_path_pattern => '$pattern', limit => 10)""".stripMargin).collect()
+ assertResult(merged.length)(unfiltered.length)
+
+ // A filter that always holds keeps every row.
+ val keepAll = spark.sql(
+ s"""call show_logfile_records(table => '$tableName',
log_file_path_pattern => '$pattern', limit => 10, filter => "records IS NOT
NULL")""".stripMargin).collect()
+ assertResult(unfiltered.length)(keepAll.length)
+
+ // A filter that never holds drops every row.
+ val dropAll = spark.sql(
+ s"""call show_logfile_records(table => '$tableName',
log_file_path_pattern => '$pattern', limit => 10, filter => "records LIKE
'%__no_such_token__%'")""".stripMargin).collect()
+ assertResult(0)(dropAll.length)
+ }
+ }
+ }
+
+ test("Test Call show_logfile_records Procedure over a non-data log block") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ // Table version 9 keeps deletes inline in the *.log.* files; under the
v10 native log default
+ // they land in separate *.deletes.* files that the log glob below would
not match.
+ createMorTable(tableName, tablePath, Some(9))
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000, 1000")
+ spark.sql(s"insert into $tableName select 2, 'a2', 20, 1500, 1000")
+ // One statement that both updates and deletes, so the delta commit
appends a data block and a
+ // delete block to the same log file. Standalone deletes are not the
only way to trip the NPE:
+ // both branches of show_logfile_records requireNonNull a schema read
off a log file, they just
+ // pick the file differently. The merged branch resolves the reader
schema once from the last
+ // globbed file; the unmerged branch re-reads it per file and dies on
the first file without a
+ // data block (or not at all, if `limit` is filled by earlier files
first). With the single-file
+ // glob asserted below, both land on the same file. The CLI twin fixed
this in HUDI-6694 (#9445)
+ // with per-file null tolerance and a backward scan for the schema; see
#19634.
+ spark.sql(
+ s"""
+ |merge into $tableName as target
+ |using (
+ | select cast(1 as int) as id, 'x1' as name, cast(10 as double) as
price,
+ | cast(2000 as long) as ts, cast(1000 as long) as partition
+ | union all
+ | select cast(2 as int) as id, 'b2' as name, cast(20 as double) as
price,
+ | cast(2000 as long) as ts, cast(1000 as long) as partition
+ |) src
+ |on target.id = src.id
+ |when matched and src.id = 1 then delete
+ |when matched and src.id = 2 then update set target.name =
src.name, target.ts = src.ts
+ |""".stripMargin)
+
+ // Known limitation: both show_logfile procedures match log blocks with
a single
+ // `case dataBlock: HoodieDataBlock` arm and no default arm, so a
non-data block -- here the
+ // delete block -- raises a MatchError instead of being skipped; see
#19634. A fix flips these
+ // assertions to assert the delete block is skipped, or surfaced,
gracefully.
+ val pattern = s"$tablePath/*/*.log.*"
+
+ // The MatchError pins below depend on small-file packing putting both
keys in ONE file group,
+ // so the data block and the delete block share a single log file. If
they ever split into two
+ // file groups, the last globbed file would be delete only and the
procedures would die with a
+ // bare NPE on the null log-file schema instead, turning this test red
for the wrong reason.
+ val globbedLogFiles = FSUtils.getGlobStatusExcludingMetaFolder(
+ createMetaClient(spark, tablePath).getStorage, new
StoragePath(pattern))
+ assertResult(1, "the delete block must share one log file with the data
block")(globbedLogFiles.size())
+
+ val deleteBlockMatchError = "of class
org.apache.hudi.common.table.log.block.HoodieDeleteBlock"
+ checkExceptionContain(
+ s"""call show_logfile_records(table => '$tableName',
log_file_path_pattern => '$pattern', limit => 10)""")(
+ deleteBlockMatchError)
+ checkExceptionContain(
+ s"""call show_logfile_metadata(table => '$tableName',
log_file_path_pattern => '$pattern', limit => 10)""")(
+ deleteBlockMatchError)
+ }
+ }
+
+ test("Test Call show_logfile_records Procedure over a delete-only log file")
{
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ // Table version 9 keeps deletes inline in the *.log.* files; under the
v10 native log default
+ // they land in separate *.deletes.* files that the log glob below would
not match.
+ createMorTable(tableName, tablePath, Some(9))
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000, 1000")
+ spark.sql(s"insert into $tableName select 2, 'a2', 20, 1500, 1000")
+ spark.sql(s"delete from $tableName where id = 1")
+
+ val pattern = s"$tablePath/*/*.log.*"
+ // The inserts go to a base file, so the only log file the glob matches
holds just the delete
+ // block -- the shape the pin below depends on.
+ val globbedLogFiles = FSUtils.getGlobStatusExcludingMetaFolder(
+ createMetaClient(spark, tablePath).getStorage, new
StoragePath(pattern))
+ assertResult(1, "the delete block must be alone in the only log
file")(globbedLogFiles.size())
+
+ // Known limitation: the unmerged path requireNonNulls the schema it
reads per log file, and a
+ // delete-only log file carries none, so the call dies with an NPE; see
#19634 crash 4. A fix
+ // -- porting HUDI-6694's per-file null tolerance and backward schema
scan -- flips this to
+ // returning the delete-block file gracefully.
+ // Assert the exception TYPE over the cause chain, message-agnostic: NPE
is in the
+ // internal-error wrap set of every Spark version, so it arrives as an
[INTERNAL_ERROR]
+ // SparkException rather than bare, and the NPE itself carries no
message of its own
+ // (Objects.requireNonNull with no message argument), with no
helpful-NPE text on JDK 11.
+ val failure = intercept[Throwable] {
+ spark.sql(
+ s"""call show_logfile_records(table => '$tableName',
log_file_path_pattern => '$pattern', limit => 10)""".stripMargin).collect()
+ }
+ assert(causeChain(failure).exists(_.isInstanceOf[NullPointerException]),
chainTypes(failure))
+ }
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala
new file mode 100644
index 000000000000..595b49a67929
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala
@@ -0,0 +1,310 @@
+/*
+ * 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.procedure
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.hudi.command.procedures.HoodieProcedureFilterUtils
+import org.apache.spark.sql.types._
+
+import java.sql.{Date, Timestamp}
+
+/**
+ * Direct unit tests for [[HoodieProcedureFilterUtils]] which evaluates SQL
filter
+ * expressions against procedure output rows. Covers
primitive/date/decimal/complex
+ * type conversion, the function-resolution table, numeric type coercion,
validation
+ * and error handling.
+ */
+class TestHoodieProcedureFilterUtils extends HoodieSparkProcedureTestBase {
+
+ private def schemaOf(fields: (String, DataType)*): StructType =
+ StructType(fields.map { case (n, dt) => StructField(n, dt, nullable =
true) })
+
+ private def keep(rows: Seq[Row], expr: String, schema: StructType): Seq[Row]
=
+ HoodieProcedureFilterUtils.evaluateFilter(rows, expr, schema, spark)
+
+ // A rich scalar schema reused across the function tests.
+ private val scalarSchema = schemaOf(
+ "id" -> IntegerType,
+ "name" -> StringType,
+ "price" -> DoubleType,
+ "ts" -> LongType,
+ "flag" -> BooleanType,
+ "neg" -> IntegerType,
+ "d" -> DateType,
+ "t" -> TimestampType)
+
+ private val scalarRows = Seq(
+ Row(1, "a1", 10.0d, 1000L, true, -5, Date.valueOf("2024-03-15"),
Timestamp.valueOf("2024-03-15 12:30:00")),
+ Row(2, "b2", 20.0d, 2000L, false, -7, Date.valueOf("2023-01-02"),
Timestamp.valueOf("2023-01-02 05:00:00")))
+
+ test("evaluateFilter returns all rows for null / blank filter") {
+ assertResult(scalarRows)(keep(scalarRows, null, scalarSchema))
+ assertResult(scalarRows)(keep(scalarRows, "", scalarSchema))
+ assertResult(scalarRows)(keep(scalarRows, " ", scalarSchema))
+ }
+
+ test("evaluateFilter handles comparison operators and boolean columns") {
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "id = 1",
scalarSchema))
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "id > 1", scalarSchema))
+ assertResult(2)(keep(scalarRows, "id >= 1", scalarSchema).length)
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "id < 2",
scalarSchema))
+ assertResult(2)(keep(scalarRows, "id <= 2", scalarSchema).length)
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "id != 1", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "name = 'a1'",
scalarSchema))
+ // The literal must match the column type, so use an explicit double
literal here; the plain
+ // 15.0 (decimal) form is pinned in the numeric-coercion test below.
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "price > 15.0d",
scalarSchema))
+ // Bare boolean column evaluates to a Boolean result directly.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "flag", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "flag = true",
scalarSchema))
+ }
+
+ test("evaluateFilter coerces Long columns against integer literals") {
+ // Exercises applyTypeCoercion for every comparison operator (Long
boundRef vs Int literal).
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts = 1000",
scalarSchema))
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "ts > 1500",
scalarSchema))
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "ts >= 2000",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts < 2000",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts <= 1000",
scalarSchema))
+ // Known limitation: the coercion narrows the Long column to Int instead
of widening the Int
+ // literal, so a Long value beyond Int range never matches (wrong results
under non-ANSI Spark,
+ // swallowed overflow error under ANSI). Pinned here so a fix flips this
assertion; see #19632.
+ val bigRow = Seq(Row(3, "c3", 30.0d, 3000000000L, true, -9,
+ Date.valueOf("2024-03-16"), Timestamp.valueOf("2024-03-16 12:30:00")))
+ assertResult(Seq.empty)(keep(bigRow, "ts > 2000", scalarSchema))
+ // Known limitation: the coercion only matches column-on-left, so a
literal-on-left comparison
+ // never coerces and drops every row instead of mirroring the equivalent
column-on-left filter.
+ // Pinned here so a fix flips these assertions; see #19632.
+ assertResult(Seq.empty)(keep(scalarRows, "1500 < ts", scalarSchema))
+ assertResult(Seq.empty)(keep(scalarRows, "1000 = ts", scalarSchema))
+ }
+
+ test("evaluateFilter does not coerce other numeric column/literal type
pairs") {
+ // Known limitation: applyTypeCoercion only special-cases a Long column
against an Int literal.
+ // Every other numeric column/literal pair is left alone, so the
mismatched comparison fails to
+ // evaluate; the per-row Try swallows the failure and drops the row. The
filter therefore
+ // returns no rows instead of erroring on the type mismatch.
+ // Pinned here so a fix flips the Seq.empty assertions; see #19632.
+ val schema = schemaOf(
+ "f" -> FloatType,
+ "sh" -> ShortType,
+ "by" -> ByteType,
+ "dec" -> DecimalType(10, 2))
+ val rows = Seq(Row(2.5f, 3.toShort, 4.toByte, new
java.math.BigDecimal("3.00")))
+
+ // Literals whose parsed type already matches the column type evaluate
correctly.
+ assertResult(rows)(keep(rows, "f > 1.0f", schema))
+ assertResult(rows)(keep(rows, "dec > 1.00", schema))
+
+ // Mismatched literal types no-match even though the values would satisfy
the predicate.
+ assertResult(Seq.empty)(keep(rows, "sh = 3", schema))
+ assertResult(Seq.empty)(keep(rows, "by = 4", schema))
+ assertResult(Seq.empty)(keep(rows, "f > 1.0d", schema))
+ assertResult(Seq.empty)(keep(rows, "dec > 1", schema))
+ // Same gap for a double column: a plain 15.0 parses as decimal, not
double.
+ assertResult(Seq.empty)(keep(scalarRows, "price > 15.0", scalarSchema))
+ }
+
+ test("evaluateFilter silently drops rows for functions outside the
resolution table") {
+ // Known limitation: a function missing from the resolution table falls
through as an
+ // UnresolvedFunction. validateFilterExpression only checks column
references, so nothing
+ // rejects it; instead evaluation fails per row and the row is dropped,
which looks like an
+ // empty result rather than an error. Pinned here so a fix flips these;
see #19638.
+ assertResult(Seq.empty)(keep(scalarRows, "concat(name, 'x') = 'a1x'",
scalarSchema))
+ assertResult(Seq.empty)(keep(scalarRows, "instr(name, 'a') = 1",
scalarSchema))
+ assertResult(Right(()))(
+ HoodieProcedureFilterUtils.validateFilterExpression("concat(name, 'x') =
'a1x'", scalarSchema, spark))
+ // if() is parsed as a function call and hits the same gap, while the
equivalent CASE WHEN is
+ // lowered by the parser without an UnresolvedFunction and evaluates fine.
+ assertResult(Seq.empty)(keep(scalarRows, "if(name = 'a1', true, false)",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(
+ keep(scalarRows, "case when name = 'a1' then true else false end",
scalarSchema))
+ // Control: a function that is in the resolution table resolves and
matches.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "upper(name) = 'A1'",
scalarSchema))
+ }
+
+ test("evaluateFilter handles AND / OR / NOT / IN / BETWEEN") {
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "id = 1 AND flag =
true", scalarSchema))
+ assertResult(2)(keep(scalarRows, "id = 1 OR id = 2", scalarSchema).length)
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "NOT (id = 1)",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "id IN (1, 3)",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "name IN ('a1', 'x')",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "id BETWEEN 1 AND 1",
scalarSchema))
+ }
+
+ test("evaluateFilter resolves string functions") {
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "upper(name) = 'A1'",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "lower(name) = 'a1'",
scalarSchema))
+ assertResult(2)(keep(scalarRows, "length(name) = 2", scalarSchema).length)
+ assertResult(2)(keep(scalarRows, "len(name) = 2", scalarSchema).length)
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "trim(name) = 'a1'",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "ltrim(name) = 'a1'",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "rtrim(name) = 'a1'",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "substr(name, 1, 1) =
'a'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "substring(name, 1, 1)
= 'a'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "name LIKE 'a%'",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "regexp_like(name,
'^a')", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "regexp_extract(name,
'([a-z]+)', 1) = 'a'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "coalesce(name, 'x') =
'a1'", scalarSchema))
+ }
+
+ test("evaluateFilter resolves numeric and cast functions") {
+ // The util only special-cases a long column vs an integer literal; every
other numeric
+ // comparison relies on the literal already matching the expression's
result type. So the
+ // literals below are typed to match: round/double yield double,
ceil/floor/long yield long.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "abs(neg) = 5",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "round(price) =
10.0d", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "round(price, 1) =
10.0d", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "ceil(price) = 10L",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "floor(price) = 10L",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "int(price) = 10",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "long(id) = 1L",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "double(id) = 1.0d",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "string(id) = '1'",
scalarSchema))
+ }
+
+ test("evaluateFilter resolves date functions") {
+ // These operate on the date column and need no session time zone, so they
evaluate directly.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "year(d) = 2024",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "month(d) = 3",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "day(d) = 15",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "dayofmonth(d) = 15",
scalarSchema))
+ assertResult(2)(keep(scalarRows, "datediff(d, d) = 0",
scalarSchema).length)
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "datediff(d,
date'2024-03-14') = 1", scalarSchema))
+ }
+
+ test("evaluateFilter cannot evaluate time-zone-aware timestamp functions") {
+ // The util binds and evaluates expressions without running Spark's
analyzer, so time-zone-aware
+ // expressions never receive a resolved time zone and fail to evaluate;
the filter then treats the
+ // row as a non-match rather than throwing. This documents that
hour()/date_format() on a timestamp
+ // are unsupported here (unlike the date functions above, which are
time-zone independent).
+ assertResult(Seq.empty)(keep(scalarRows, "hour(t) = 12", scalarSchema))
+ assertResult(Seq.empty)(keep(scalarRows, "date_format(t, 'yyyy') =
'2024'", scalarSchema))
+ }
+
+ test("evaluateFilter maps a string result of true / false to a boolean
decision") {
+ // string(flag) yields the literal strings "true"/"false", exercising the
string->boolean branch.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "string(flag)",
scalarSchema))
+ }
+
+ test("evaluateFilter treats non-boolean-valued expressions as no-match") {
+ // A bare string/int column produces a value that is neither Boolean nor
"true"/"false".
+ assertResult(Seq.empty)(keep(scalarRows, "name", scalarSchema))
+ assertResult(Seq.empty)(keep(scalarRows, "id", scalarSchema))
+ }
+
+ test("evaluateFilter handles null values and IS [NOT] NULL") {
+ val schema = schemaOf("id" -> IntegerType, "name" -> StringType)
+ val rows = Seq(Row(1, "a1"), Row(2, null))
+ assertResult(Seq(rows(1)))(keep(rows, "isnull(name)", schema))
+ assertResult(Seq(rows.head))(keep(rows, "isnotnull(name)", schema))
+ assertResult(Seq(rows(1)))(keep(rows, "name IS NULL", schema))
+ assertResult(Seq(rows.head))(keep(rows, "name IS NOT NULL", schema))
+ // A null-valued bare expression resolves to a false decision.
+ assertResult(Seq.empty)(keep(Seq(rows(1)), "name", schema))
+ }
+
+ test("evaluateFilter converts map columns and resolves map functions") {
+ import scala.collection.JavaConverters._
+ val schema = schemaOf("id" -> IntegerType,
+ "mScala" -> MapType(StringType, IntegerType),
+ "mJava" -> MapType(StringType, IntegerType))
+ val javaMap = Map("a" -> 1, "b" -> 2).asJava
+ val rows = Seq(Row(1, Map("a" -> 1, "b" -> 2), javaMap))
+ assertResult(rows)(keep(rows, "size(mScala) = 2", schema))
+ assertResult(rows)(keep(rows, "size(mJava) = 2", schema))
+ assertResult(rows)(keep(rows, "array_contains(map_keys(mScala), 'a')",
schema))
+ assertResult(rows)(keep(rows, "array_contains(map_values(mScala), 1)",
schema))
+ assertResult(Seq.empty)(keep(rows, "size(mScala) = 5", schema))
+ }
+
+ test("evaluateFilter converts struct columns and resolves IS [NOT] NULL on
them") {
+ val schema = schemaOf("id" -> IntegerType,
+ "st" -> StructType(Seq(StructField("x", IntegerType), StructField("y",
StringType))))
+ val rows = Seq(Row(1, Row(10, "p")), Row(2, null))
+ assertResult(Seq(rows.head))(keep(rows, "isnotnull(st)", schema))
+ assertResult(Seq(rows(1)))(keep(rows, "isnull(st)", schema))
+ }
+
+ test("evaluateFilter converts array / decimal / binary / uuid / java-time
columns without error") {
+ import scala.collection.JavaConverters._
+ val schema = schemaOf(
+ "id" -> IntegerType,
+ "arrScala" -> ArrayType(IntegerType),
+ "arrList" -> ArrayType(IntegerType),
+ "arrArray" -> ArrayType(IntegerType),
+ "dec" -> DecimalType(10, 2),
+ "decScala" -> DecimalType(10, 2),
+ "bin" -> BinaryType,
+ "uuidCol" -> StringType,
+ "inst" -> TimestampType,
+ "ld" -> DateType,
+ "ldt" -> TimestampType)
+ val row = Row(
+ 1,
+ Seq(1, 2, 3),
+ List(1, 2, 3).map(Int.box).asJava,
+ Array(1, 2, 3),
+ new java.math.BigDecimal("12.50"),
+ scala.math.BigDecimal("34.75"),
+ Array[Byte](1, 2, 3),
+ java.util.UUID.randomUUID(),
+ java.time.Instant.parse("2024-03-15T12:30:00Z"),
+ java.time.LocalDate.of(2024, 3, 15),
+ java.time.LocalDateTime.of(2024, 3, 15, 12, 30, 0))
+ val rows = Seq(row)
+ // Filtering on the scalar column converts every field of the row,
exercising each
+ // complex-type conversion branch; the row is retained.
+ assertResult(rows)(keep(rows, "id = 1", schema))
+ // Decimal / binary / uuid / java-time values survive conversion and are
non-null.
+ assertResult(rows)(keep(rows, "isnotnull(dec) AND isnotnull(decScala)",
schema))
+ assertResult(rows)(keep(rows, "isnotnull(bin) AND isnotnull(uuidCol)",
schema))
+ assertResult(rows)(keep(rows, "isnotnull(inst) AND isnotnull(ld) AND
isnotnull(ldt)", schema))
+ // Known limitation: array values are converted to a plain Array instead
of Catalyst ArrayData,
+ // so every array predicate (even isnotnull) fails to evaluate and drops
the row instead of
+ // matching. Pinned here so a fix flips these assertions; see #19633.
+ assertResult(Seq.empty)(keep(rows, "size(arrScala) >= 0", schema))
+ assertResult(Seq.empty)(keep(rows, "isnotnull(arrScala)", schema))
+ }
+
+ test("evaluateFilter throws IllegalArgumentException on unparseable
expression") {
+ val e = intercept[IllegalArgumentException] {
+ keep(scalarRows, "id >< 1", scalarSchema)
+ }
+ assert(e.getMessage.contains("Failed to parse or evaluate filter
expression"))
+ }
+
+ test("validateFilterExpression accepts valid references and rejects unknown
ones") {
+ assertResult(Right(()))(
+ HoodieProcedureFilterUtils.validateFilterExpression("id > 1 AND name =
'a1'", scalarSchema, spark))
+ assertResult(Right(()))(
+ HoodieProcedureFilterUtils.validateFilterExpression(null, scalarSchema,
spark))
+ assertResult(Right(()))(
+ HoodieProcedureFilterUtils.validateFilterExpression(" ", scalarSchema,
spark))
+
+ val invalidCol =
HoodieProcedureFilterUtils.validateFilterExpression("missing_col > 1",
scalarSchema, spark)
+ assert(invalidCol.isLeft)
+ val invalidColMsg = invalidCol.fold(identity, _ => "")
+ assert(invalidColMsg.contains("Invalid column references"))
+ assert(invalidColMsg.contains("missing_col"))
+
+ val parseError = HoodieProcedureFilterUtils.validateFilterExpression("id
>< 1", scalarSchema, spark)
+ assert(parseError.isLeft)
+ assert(parseError.fold(identity, _ => "").contains("Invalid filter
expression"))
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestProcedureParameterImpl.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestProcedureParameterImpl.scala
new file mode 100644
index 000000000000..fac0db989444
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestProcedureParameterImpl.scala
@@ -0,0 +1,63 @@
+/*
+ * 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.procedure
+
+import org.apache.spark.sql.hudi.command.procedures.ProcedureParameter
+import org.apache.spark.sql.types.DataTypes
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse,
assertNotEquals, assertTrue}
+import org.scalatest.funsuite.AnyFunSuite
+
+/**
+ * Unit tests for ProcedureParameterImpl.equals. The implementation used to
call this == other for
+ * the identity check, which dispatches straight back into equals and recurses
until it overflows
+ * the stack, and it cast the argument before the null/type guard, so a
foreign argument threw
+ * ClassCastException. Both are self-contained checks, so no Spark session is
needed.
+ */
+class TestProcedureParameterImpl extends AnyFunSuite {
+
+ test("Test ProcedureParameterImpl equals identity, foreign types and null") {
+ val param = ProcedureParameter.optional(0, "table", DataTypes.StringType,
"default")
+
+ // Self comparison must terminate instead of recursing into equals.
+ assertTrue(param == param)
+ assertTrue(param.equals(param))
+
+ // A foreign type and null are rejected by the guard rather than blowing
up on the cast.
+ assertFalse(param.equals("notAParam"))
+ assertFalse(param.equals(null))
+ }
+
+ test("Test ProcedureParameterImpl equals and hashCode over field values") {
+ val required = ProcedureParameter.required(2, "instant",
DataTypes.StringType)
+ val sameAsRequired = ProcedureParameter.required(2, "instant",
DataTypes.StringType)
+ assertEquals(required, sameAsRequired)
+ assertEquals(required.hashCode(), sameAsRequired.hashCode())
+
+ val optional = ProcedureParameter.optional(1, "dry_run",
DataTypes.BooleanType, true)
+ val sameAsOptional = ProcedureParameter.optional(1, "dry_run",
DataTypes.BooleanType, true)
+ assertEquals(optional, sameAsOptional)
+ assertEquals(optional.hashCode(), sameAsOptional.hashCode())
+
+ // One differing field at a time.
+ assertNotEquals(optional, ProcedureParameter.optional(2, "dry_run",
DataTypes.BooleanType, true))
+ assertNotEquals(optional, ProcedureParameter.optional(1, "backup",
DataTypes.BooleanType, true))
+ assertNotEquals(optional, ProcedureParameter.optional(1, "dry_run",
DataTypes.StringType, true))
+ assertNotEquals(optional, ProcedureParameter.optional(1, "dry_run",
DataTypes.BooleanType, false))
+ assertNotEquals(optional, ProcedureParameter.required(1, "dry_run",
DataTypes.BooleanType))
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairClusteringPlanProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairClusteringPlanProcedure.scala
index 9c1d87904b4d..bdf75b688eca 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairClusteringPlanProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairClusteringPlanProcedure.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.hudi.procedure
import org.apache.hudi.avro.model.HoodieClusteringPlan
-import org.apache.hudi.common.table.timeline.HoodieTimeline
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.hudi.common.util.ClusteringUtils
import org.apache.hudi.storage.StoragePath
import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
@@ -36,35 +36,7 @@ class TestRepairClusteringPlanProcedure extends
HoodieSparkProcedureTestBase {
withSQLConf("hoodie.parquet.small.file.limit" -> "0") {
val tableName = generateTableName
val basePath = s"${tmp.getCanonicalPath}/$tableName"
-
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long,
- | dt string
- |) using hudi
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | orderingFields = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- | partitioned by (dt)
- | location '$basePath'
- |""".stripMargin)
-
- insertRows(tableName, "2026-01-27", 1)
- insertRows(tableName, "2026-01-27", 4)
- insertRows(tableName, "2026-01-27", 7)
-
- runClustering(tableName, "schedule")
- var metadata = getLatestRequestedClusteringPlan(basePath)
- val instant1 = metadata.getLeft
- val plannedFiles = getPlanDataFiles(metadata.getRight)
- assertTrue(plannedFiles.size > 1, s"Expected more than one planned
file, but got $plannedFiles")
+ val (instant1, plannedFiles) =
createTableWithPendingClusteringPlan(tableName, basePath)
val fileToRemove = plannedFiles.head
val dryRunRows = spark.sql(
@@ -78,8 +50,7 @@ class TestRepairClusteringPlanProcedure extends
HoodieSparkProcedureTestBase {
s"op => 'delete', invalid_parquet_files => '$fileToRemove',
dry_run => false)").collect()
assertRepairResult(deleteRows, instant1.requestedTime, fileToRemove,
"REMOVED_FROM_PLAN", deleted = false, "USER_REQUESTED")
- metadata = getLatestRequestedClusteringPlan(basePath)
- assertFalse(getPlanDataFiles(metadata.getRight).contains(fileToRemove))
+
assertFalse(getPlanDataFiles(getLatestRequestedClusteringPlan(basePath).getRight).contains(fileToRemove))
var metaClient = createMetaClient(spark, basePath)
assertTrue(metaClient.getStorage.exists(new StoragePath(fileToRemove)))
@@ -99,9 +70,9 @@ class TestRepairClusteringPlanProcedure extends
HoodieSparkProcedureTestBase {
insertRows(tableName, "2026-01-28", 16)
runClustering(tableName, "schedule")
- metadata = getLatestRequestedClusteringPlan(basePath)
- val instant2 = metadata.getLeft
- val corruptFile = getPlanDataFiles(metadata.getRight).head
+ val metadata2 = getLatestRequestedClusteringPlan(basePath)
+ val instant2 = metadata2.getLeft
+ val corruptFile = getPlanDataFiles(metadata2.getRight).head
metaClient = createMetaClient(spark, basePath)
replaceWithEmptyFile(metaClient.getStorage, new
StoragePath(corruptFile))
@@ -110,8 +81,7 @@ class TestRepairClusteringPlanProcedure extends
HoodieSparkProcedureTestBase {
s"op => 'validate_delete', need_delete => true, dry_run =>
false)").collect()
assertRepairResult(validateRows, instant2.requestedTime, corruptFile,
"REMOVED_FROM_PLAN", deleted = true, "NOT_PARQUET_FILE")
- metadata = getLatestRequestedClusteringPlan(basePath)
- assertFalse(getPlanDataFiles(metadata.getRight).contains(corruptFile))
+
assertFalse(getPlanDataFiles(getLatestRequestedClusteringPlan(basePath).getRight).contains(corruptFile))
metaClient = createMetaClient(spark, basePath)
assertFalse(metaClient.getStorage.exists(new StoragePath(corruptFile)))
}
@@ -122,6 +92,126 @@ class TestRepairClusteringPlanProcedure extends
HoodieSparkProcedureTestBase {
assertNotNull(org.apache.spark.sql.hudi.command.procedures.HoodieProcedures.newBuilder("repair_clustering_plan"))
}
+ test("Test repair_clustering_plan validate_delete, unsupported op and
allow_empty_plan guard") {
+ withTempDir { tmp =>
+ withSQLConf("hoodie.parquet.small.file.limit" -> "0") {
+ val tableName = generateTableName
+ val basePath = s"${tmp.getCanonicalPath}/$tableName"
+ val (instant, plannedFiles) =
createTableWithPendingClusteringPlan(tableName, basePath)
+
+ // validate_delete over a healthy plan finds no corrupt files, so no
candidates are returned.
+ // Anchor the negative result: every planned file exists and is
readable on storage.
+ val metaClient = createMetaClient(spark, basePath)
+ plannedFiles.foreach(file =>
assertTrue(metaClient.getStorage.exists(new StoragePath(file)),
+ s"Expected planned file $file to exist"))
+ val healthyRows = spark.sql(
+ s"call repair_clustering_plan(table => '$tableName', instant =>
'${instant.requestedTime}', " +
+ s"op => 'validate_delete')").collect()
+ assertEquals(0, healthyRows.length)
+
+ // An unknown operation is rejected.
+ checkExceptionContain(
+ s"call repair_clustering_plan(table => '$tableName', instant =>
'${instant.requestedTime}', op => 'bogus')")(
+ "Unsupported operation")
+
+ // A candidate that is not part of the plan is reported as
NOT_FOUND_IN_PLAN and, because
+ // deletion only applies to the (candidates intersect plan) set, the
file is never deleted
+ // even with need_delete => true.
+ insertRows(tableName, "2026-01-28", 10)
+ val foreign = spark.sql(
+ s"select distinct _hoodie_partition_path, _hoodie_file_name from
$tableName where dt = '2026-01-28'")
+ .collect().head
+ val notInPlanFile =
s"$basePath/${foreign.getString(0)}/${foreign.getString(1)}"
+ val notInPlanRows = spark.sql(
+ s"call repair_clustering_plan(table => '$tableName', instant =>
'${instant.requestedTime}', " +
+ s"op => 'delete', invalid_parquet_files => '$notInPlanFile',
need_delete => true, dry_run => false)").collect()
+ assertRepairResult(notInPlanRows, instant.requestedTime,
notInPlanFile, "NOT_FOUND_IN_PLAN", deleted = false, "USER_REQUESTED")
+ assertTrue(metaClient.getStorage.exists(new
StoragePath(notInPlanFile)),
+ s"Expected $notInPlanFile to survive a repair request it is not
planned in")
+ assertEquals(plannedFiles.toSet,
getPlanDataFiles(getLatestRequestedClusteringPlan(basePath).getRight).toSet)
+
+ // Removing every input group without allow_empty_plan is rejected.
+ val allFilesArg = plannedFiles.distinct.mkString(",")
+ checkExceptionContain(
+ s"call repair_clustering_plan(table => '$tableName', instant =>
'${instant.requestedTime}', " +
+ s"op => 'delete', invalid_parquet_files => '$allFilesArg')")(
+ "would remove all input groups")
+
+ // With allow_empty_plan the destructive run removes every planned
file and rewrites the
+ // requested instant with an empty plan that still parses.
+ val emptyPlanRows = spark.sql(
+ s"call repair_clustering_plan(table => '$tableName', instant =>
'${instant.requestedTime}', " +
+ s"op => 'delete', invalid_parquet_files => '$allFilesArg',
allow_empty_plan => true, dry_run => false)").collect()
+ assertEquals(plannedFiles.distinct.size, emptyPlanRows.length)
+ assertTrue(emptyPlanRows.forall(_.getString(2) == "REMOVED_FROM_PLAN"),
+ s"Expected all rows to be REMOVED_FROM_PLAN, but got
${emptyPlanRows.mkString("[", ", ", "]")}")
+ val emptiedPlan = getLatestRequestedClusteringPlan(basePath)
+ assertEquals(instant.requestedTime, emptiedPlan.getLeft.requestedTime)
+ assertTrue(getPlanDataFiles(emptiedPlan.getRight).isEmpty,
+ s"Expected the rewritten plan to have no data files, but got
${getPlanDataFiles(emptiedPlan.getRight)}")
+ }
+ }
+ }
+
+ test("Test repair_clustering_plan with backup disabled") {
+ withTempDir { tmp =>
+ withSQLConf("hoodie.parquet.small.file.limit" -> "0") {
+ val tableName = generateTableName
+ val basePath = s"${tmp.getCanonicalPath}/$tableName"
+ val (instant, plannedFiles) =
createTableWithPendingClusteringPlan(tableName, basePath)
+
+ // With backup => false the requested instant file is deleted and
rewritten in place with no
+ // copy taken first, which is a distinct arm from the default backup
=> true path the other
+ // tests cover. The restore-on-failure arm (a backed up instant
restored after the rewrite
+ // throws) remains uncovered, since it needs an injected storage
failure.
+ val fileToRemove = plannedFiles.head
+ val rows = spark.sql(
+ s"call repair_clustering_plan(table => '$tableName', instant =>
'${instant.requestedTime}', " +
+ s"op => 'delete', invalid_parquet_files => '$fileToRemove', backup
=> false, dry_run => false)").collect()
+ assertRepairResult(rows, instant.requestedTime, fileToRemove,
"REMOVED_FROM_PLAN", deleted = false, "USER_REQUESTED")
+
+ val repairedPlan = getLatestRequestedClusteringPlan(basePath)
+ assertEquals(instant.requestedTime, repairedPlan.getLeft.requestedTime)
+
assertFalse(getPlanDataFiles(repairedPlan.getRight).contains(fileToRemove))
+
+ val metaClient = createMetaClient(spark, basePath)
+ assertFalse(metaClient.getStorage.exists(new
StoragePath(metaClient.getMetaPath, ".repair")),
+ "Expected no backup under .repair when backup => false")
+ }
+ }
+ }
+
+ private def createTableWithPendingClusteringPlan(tableName: String,
basePath: String): (HoodieInstant, Seq[String]) = {
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long,
+ | dt string
+ |) using hudi
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | orderingFields = 'ts',
+ | hoodie.metadata.enable = 'false'
+ | )
+ | partitioned by (dt)
+ | location '$basePath'
+ |""".stripMargin)
+
+ insertRows(tableName, "2026-01-27", 1)
+ insertRows(tableName, "2026-01-27", 4)
+ insertRows(tableName, "2026-01-27", 7)
+
+ runClustering(tableName, "schedule")
+ val metadata = getLatestRequestedClusteringPlan(basePath)
+ val plannedFiles = getPlanDataFiles(metadata.getRight)
+ assertTrue(plannedFiles.size > 1, s"Expected more than one planned file,
but got $plannedFiles")
+ (metadata.getLeft, plannedFiles)
+ }
+
private def runClustering(tableName: String,
operation: String,
instant: Option[String] = None): Unit = {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala
index c35d96dacfd2..9fac4df2f072 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala
@@ -18,6 +18,11 @@
package org.apache.spark.sql.hudi.procedure
import org.apache.hudi.HoodieSparkUtils
+import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
+
+import java.io.IOException
+
+import scala.collection.JavaConverters._
class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
@@ -213,78 +218,6 @@ class TestShowCleansProcedures extends
HoodieSparkProcedureTestBase {
}
}
- test("Test show procedures with showArchived parameter") {
- withTempDir { tmp =>
- Seq("COPY_ON_WRITE").foreach { tableType =>
- val tableName = generateTableName
- val tablePath = s"${tmp.getCanonicalPath}/$tableName"
- val extraConf = if (HoodieSparkUtils.gteqSpark3_4) {
- Map("spark.sql.defaultColumn.enabled" -> "false")
- } else {
- Map.empty[String, String]
- }
- withSQLConf(extraConf.toSeq: _*) {
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = '$tableType',
- | preCombineField = 'ts'
- | )
- |""".stripMargin)
-
- spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
- spark.sql(s"insert into $tableName values(2, 'a2', 20, 2000)")
- spark.sql(s"update $tableName set price = 11 where id = 1")
-
- spark.sql(s"call run_clean(table => '$tableName', retain_commits =>
1)")
- .collect()
-
- // showArchived=false (default - active timeline only)
- val activeCleans = spark.sql(s"call show_cleans(table =>
'$tableName', showArchived => false)")
- .collect()
- spark.sql(s"call show_cleans(table => '$tableName', showArchived =>
false)").show(false)
-
- val activePlans = spark.sql(s"call show_clean_plans(table =>
'$tableName', showArchived => false)")
- .collect()
- spark.sql(s"call show_clean_plans(table => '$tableName',
showArchived => false)").show(false)
-
- val activeMetadata = spark.sql(s"call show_cleans_metadata(table =>
'$tableName', showArchived => false)")
- .collect()
- spark.sql(s"call show_cleans_metadata(table => '$tableName',
showArchived => false)").show(false)
-
- // showArchived=true (both active + archived timelines merged)
- val allCleans = spark.sql(s"call show_cleans(table => '$tableName',
showArchived => true)")
- .collect()
- spark.sql(s"call show_cleans(table => '$tableName', showArchived =>
true)").show(false)
-
- val allPlans = spark.sql(s"call show_clean_plans(table =>
'$tableName', showArchived => true)")
- .collect()
- spark.sql(s"call show_clean_plans(table => '$tableName',
showArchived => true)").show(false)
-
- val allMetadata = spark.sql(s"call show_cleans_metadata(table =>
'$tableName', showArchived => true)")
- .collect()
- spark.sql(s"call show_cleans_metadata(table => '$tableName',
showArchived => true)").show(false)
-
- assert(activeCleans.length >= 1, "Active timeline should have clean
instances")
- assert(activePlans.length >= 1, "Active timeline should have clean
plans")
-
- // showArchived=true should include at least the same data as active
timeline
- assert(allCleans.length >= activeCleans.length, "Active + Archived
should have at least as many instances as active only")
- assert(allPlans.length >= activePlans.length, "Active + Archived
should have at least as many plans as active only")
- assert(allMetadata.length >= activeMetadata.length, "Active +
Archived should have at least as much metadata as active only")
- }
- }
- }
- }
-
test("Test show_cleans procedures with limit parameter") {
withSQLConf("hoodie.clean.automatic" -> "false",
"hoodie.parquet.max.file.size" -> "10000") {
withTempDir { tmp =>
@@ -547,4 +480,163 @@ class TestShowCleansProcedures extends
HoodieSparkProcedureTestBase {
}
}
}
+
+ test("Test show_clean_plans with an archived clean instant") {
+ withSQLConf("hoodie.clean.automatic" -> "false",
"hoodie.archive.automatic" -> "false") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ | ) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.metadata.enable = 'false'
+ | )
+ |""".stripMargin)
+
+ def rowCount(procedure: String, showArchived: Boolean): Int =
+ spark.sql(s"call $procedure(table => '$tableName', showArchived =>
$showArchived)").collect().length
+
+ // Six write commits with a clean in the middle: the first clean sits
before the last
+ // commit that archival will move, so it gets archived along with
those commits, while
+ // the second clean stays on the active timeline.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ spark.sql(s"update $tableName set price = 11 where id = 1")
+ spark.sql(s"update $tableName set price = 12 where id = 1")
+ spark.sql(s"call run_clean(table => '$tableName', retain_commits =>
1)").collect()
+ spark.sql(s"update $tableName set price = 13 where id = 1")
+ spark.sql(s"update $tableName set price = 14 where id = 1")
+ spark.sql(s"update $tableName set price = 15 where id = 1")
+ spark.sql(s"call run_clean(table => '$tableName', retain_commits =>
1)").collect()
+
+ // Nothing has been archived yet, so showArchived => true here only
exercises the merge of
+ // the active timeline with an empty archived one. All three
procedures succeed and see
+ // both cleans, which is what makes the failures asserted after
archival attributable to
+ // the archived instants themselves rather than to the merged read
path. Two rows for
+ // show_cleans_metadata as well, since that procedure emits one row
per partition per
+ // clean and this table is not partitioned.
+ assertResult(2)(rowCount("show_cleans", showArchived = true))
+ assertResult(2)(rowCount("show_cleans_metadata", showArchived = true))
+ assertResult(2)(rowCount("show_clean_plans", showArchived = true))
+
+ spark.sql(s"call archive_commits(table => '$tableName', min_commits =>
2, max_commits => 3," +
+ " retain_commits => 1, enable_metadata => false)").collect()
+
+ // Precondition: archival must have split the two cleans across the
two timelines,
+ // otherwise the archived branch below is never exercised and the test
passes vacuously.
+ val metaClient = createMetaClient(spark, tablePath)
+ val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline
+ .getInstants.asScala.map(_.requestedTime).toSeq
+ val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline
+ .getInstants.asScala.map(_.requestedTime).toSeq
+ assert(archivedCleans.length == 1, s"expected exactly 1 archived
clean, got $archivedCleans")
+ assert(activeCleans.length == 1, s"expected exactly 1 active clean,
got $activeCleans")
+ assert(archivedCleans.head.compareTo(activeCleans.head) < 0,
+ "the archived clean must be the older of the two")
+
+ // Sibling-procedure controls, pinning what the other two procedures
do with the very same
+ // archived clean. On the active timeline all three agree: one row for
the one active
+ // clean. With showArchived => true they diverge, and neither sibling
is correct today.
+ assertResult(1)(rowCount("show_cleans", showArchived = false))
+ assertResult(1)(rowCount("show_cleans_metadata", showArchived = false))
+ assertResult(1)(rowCount("show_clean_plans", showArchived = false))
+
+ // The other half of #19639, which covers all three clean procedures.
show_cleans and its
+ // show_cleans_metadata variant do route to getArchivedTimeline, but
the archived instants
+ // carry no content there, so readCleanMetadata cannot deserialize
them and the call fails
+ // outright rather than degrading to a partial row. Same
missing-archived-content cause as
+ // the all-null plan rows asserted below, just a harsher symptom.
Pinned as observed.
+ Seq("show_cleans", "show_cleans_metadata").foreach { procedure =>
+ val e = intercept[IOException](rowCount(procedure, showArchived =
true))
+ assert(e.getMessage.contains(archivedCleans.head),
+ s"$procedure over the archived timeline should fail on the
archived clean" +
+ s" ${archivedCleans.head}, but failed with: ${e.getMessage}")
+ }
+
+ val plans = spark.sql(s"call show_clean_plans(table => '$tableName',
showArchived => true)").collect()
+ assertResult(2)(plans.length)
+ val planTimes = plans.map(_.getString(0)).mkString(", ")
+ val activePlan = plans.find(_.getString(0) == activeCleans.head)
+ .getOrElse(fail(s"no plan row for the active clean
${activeCleans.head}, got plan_times: $planTimes"))
+ val archivedPlan = plans.find(_.getString(0) == archivedCleans.head)
+ .getOrElse(fail(s"no plan row for the archived clean
${archivedCleans.head}, got plan_times: $planTimes"))
+
+ // Fields that come from the cleaner plan itself, resolved by name off
the row schema so
+ // that a change in output-schema ordering cannot silently re-point
these assertions.
+ // extra_metadata is left out: it is null for both rows, so it does
not discriminate.
+ val planFields = Seq(
+ "earliest_instant_to_retain",
+ "last_completed_commit_timestamp",
+ "policy",
+ "version",
+ "total_partitions_to_clean",
+ "total_partitions_to_delete")
+
+ // The active clean plan is read correctly.
+ assertResult("COMPLETED")(activePlan.getString(1))
+ assertResult("clean")(activePlan.getString(2))
+ planFields.foreach { name =>
+ assert(!activePlan.isNullAt(activePlan.fieldIndex(name)), s"active
clean plan should have a non-null $name")
+ }
+
+ // Known limitation, see #19639: getCleanerPlans collects the archived
clean instants but
+ // then hands every instant to processCleanPlan against the ACTIVE
timeline, so the
+ // archived instant's .clean.requested file is not found, the read
falls back to
+ // createErrorRow and every plan field comes back null. Only
plan_time/state/action
+ // survive, because those are taken from the instant and not from the
plan. A fix has to
+ // read the archived instant's own content and would flip the null
assertions below to
+ // non-null; merely routing to getArchivedTimeline the way the sibling
ShowCleansProcedure
+ // does is not enough, since that path throws today (pinned above).
+ assertResult("COMPLETED")(archivedPlan.getString(1))
+ assertResult("clean")(archivedPlan.getString(2))
+ planFields.foreach { name =>
+ assert(archivedPlan.isNullAt(archivedPlan.fieldIndex(name)),
+ s"archived clean plan is expected to return a null $name today
(#19639)")
+ }
+ }
+ }
+ }
+
+ test("Test show_clean_plans validates its inputs") {
+ withSQLConf("hoodie.clean.automatic" -> "false") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ | ) using hudi
+ | location '${tmp.getCanonicalPath}'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts'
+ | )
+ |""".stripMargin)
+
+ // No insert needed: both validations run before any table data is
read.
+ // A non-positive limit is rejected.
+ checkExceptionContain(s"call show_clean_plans(table => '$tableName',
limit => 0)")(
+ "Limit must be positive")
+
+ // A filter that references an unknown column is rejected before the
plans are read; the
+ // column-reference message discriminates this branch from a filter
parse failure.
+ checkExceptionContain(
+ s"""call show_clean_plans(table => '$tableName', filter =>
"nonexistent_col > 1")""")(
+ "Invalid column references: nonexistent_col")
+ }
+ }
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestValidateHoodieSyncProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestValidateHoodieSyncProcedure.scala
new file mode 100644
index 000000000000..2cc640e7e971
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestValidateHoodieSyncProcedure.scala
@@ -0,0 +1,254 @@
+/*
+ * 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.procedure
+
+import org.apache.hudi.exception.HoodieIOException
+
+import java.sql.SQLException
+
+/**
+ * Tests for
[[org.apache.spark.sql.hudi.command.procedures.ValidateHoodieSyncProcedure]].
+ *
+ * The "complete" / "latestPartitions" modes count records over JDBC, but they
do not need a live
+ * Hive/JDBC endpoint to be exercised negatively: pointed at a portless URL,
both fail fast in the
+ * driver's transport setup, which rejects the invalid port before any network
connect. The last
+ * test pins those two failure shapes, one of which is the connection-failure
masking bug of #19635.
+ *
+ * The other tests pass 'noop', which short-circuits the record counting
(record counts stay 0)
+ * while still exercising the timeline comparison, the catch-up-commit
computation and the result
+ * formatting, which is the bulk of the procedure.
+ */
+class TestValidateHoodieSyncProcedure extends HoodieSparkProcedureTestBase {
+
+ /**
+ * The throwable and its causes, in order. Null-terminated, and capped so a
self-referential
+ * cause cannot spin forever.
+ */
+ private def causeChain(t: Throwable): Seq[Throwable] =
+ Iterator.iterate(t)(_.getCause).takeWhile(_ != null).take(16).toSeq
+
+ private def chainTypes(t: Throwable): String =
causeChain(t).map(_.getClass.getName).mkString(" <- ")
+
+ private def createTable(tableName: String, path: String, tableType: String =
"cow"): Unit = {
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | location '$path'
+ | tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ | )
+ """.stripMargin)
+ }
+
+ test("Test Call sync_validate when the target table is ahead (catch-up
commits)") {
+ withTempDir { tmp =>
+ // Each insert must land in its own file group: the catch-up count is
+ // recordsWritten - updateRecordsWritten summed over the catch-up
commits, so without this
+ // conf the second insert packs into the first small file, rewrites the
copied first record
+ // and the count becomes 3 instead of 2.
+ withSQLConf("hoodie.parquet.small.file.limit" -> "0") {
+ val srcTable = generateTableName
+ val dstTable = generateTableName
+ createTable(srcTable, s"${tmp.getCanonicalPath}/$srcTable")
+ // The source table has a single, earlier commit.
+ spark.sql(s"insert into $srcTable select 1, 'a1', 10, 1000")
+
+ createTable(dstTable, s"${tmp.getCanonicalPath}/$dstTable")
+ // The destination table receives later commits, so it is ahead of the
source.
+ spark.sql(s"insert into $dstTable select 1, 'a1', 10, 1000")
+ spark.sql(s"insert into $dstTable select 2, 'a2', 20, 2000")
+
+ val result = spark.sql(
+ s"""call sync_validate(src_table => '$srcTable', dst_table =>
'$dstTable',
+ | mode => 'noop', hive_server_url => 'jdbc:hive2://unused',
hive_pass => 'x')"""
+ .stripMargin).collect()
+
+ assertResult(1)(result.length)
+ // The procedure prints the unqualified table name, while
generateTableName is db-qualified.
+ val srcName = srcTable.stripPrefix("default.")
+ val dstName = dstTable.stripPrefix("default.")
+ // The destination is ahead, so the dst-first branch is taken
(count(dst) - count(src)) and the
+ // two catch-up commits (one insert record each) are counted. Record
counts stay 0 in this mode.
+ // "Catach up" mirrors the typo in the procedure's output message.
+ assertResult(s"Count difference now is count($dstName) -
count($srcName) == 0. Catach up count is 2")(
+ result.head.getString(0))
+ }
+ }
+ }
+
+ test("Test Call sync_validate when both tables point at the same timeline
(no catch-up)") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ createTable(tableName, s"${tmp.getCanonicalPath}/$tableName")
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+
+ // Using the same table as both source and target means neither timeline
is ahead,
+ // so no catch-up commits are found and only the count difference is
reported.
+ val result = spark.sql(
+ s"""call sync_validate(src_table => '$tableName', dst_table =>
'$tableName',
+ | mode => 'noop', hive_server_url => 'jdbc:hive2://unused',
hive_pass => 'x')"""
+ .stripMargin).collect()
+
+ assertResult(1)(result.length)
+ // The procedure prints the unqualified table name, while
generateTableName is db-qualified.
+ val name = tableName.stripPrefix("default.")
+ // No catch-up suffix: the exact match pins that no commits are found
after the (shared) latest.
+ assertResult(s"Count difference now is count($name) - count($name) ==
0")(
+ result.head.getString(0))
+ }
+ }
+
+ test("Test Call sync_validate when the source table is ahead (argument
order)") {
+ withTempDir { tmp =>
+ withSQLConf("hoodie.parquet.small.file.limit" -> "0") {
+ val srcTable = generateTableName
+ val dstTable = generateTableName
+ createTable(dstTable, s"${tmp.getCanonicalPath}/$dstTable")
+ spark.sql(s"insert into $dstTable select 1, 'a1', 10, 1000")
+
+ createTable(srcTable, s"${tmp.getCanonicalPath}/$srcTable")
+ spark.sql(s"insert into $srcTable select 1, 'a1', 10, 1000")
+ spark.sql(s"insert into $srcTable select 2, 'a2', 20, 2000")
+
+ val result = spark.sql(
+ s"""call sync_validate(src_table => '$srcTable', dst_table =>
'$dstTable',
+ | mode => 'noop', hive_server_url => 'jdbc:hive2://unused',
hive_pass => 'x')"""
+ .stripMargin).collect()
+
+ assertResult(1)(result.length)
+ // The procedure prints the unqualified table name, while
generateTableName is db-qualified.
+ val srcName = srcTable.stripPrefix("default.")
+ val dstName = dstTable.stripPrefix("default.")
+ // The source is ahead, so the else branch is taken: this pins that it
passes the source
+ // table first, i.e. count(src) - count(dst), the mirror image of the
dst-ahead branch.
+ assertResult(s"Count difference now is count($srcName) -
count($dstName) == 0. Catach up count is 2")(
+ result.head.getString(0))
+ }
+ }
+ }
+
+ test("Test Call sync_validate when the target table has never been written
(sentinel commit)") {
+ withTempDir { tmp =>
+ withSQLConf("hoodie.parquet.small.file.limit" -> "0") {
+ val srcTable = generateTableName
+ val dstTable = generateTableName
+ createTable(srcTable, s"${tmp.getCanonicalPath}/$srcTable")
+ spark.sql(s"insert into $srcTable select 1, 'a1', 10, 1000")
+ spark.sql(s"insert into $srcTable select 2, 'a2', 20, 2000")
+
+ // The destination is created but never written, so its commits
timeline is empty.
+ createTable(dstTable, s"${tmp.getCanonicalPath}/$dstTable")
+
+ val result = spark.sql(
+ s"""call sync_validate(src_table => '$srcTable', dst_table =>
'$dstTable',
+ | mode => 'noop', hive_server_url => 'jdbc:hive2://unused',
hive_pass => 'x')"""
+ .stripMargin).collect()
+
+ assertResult(1)(result.length)
+ // The procedure prints the unqualified table name, while
generateTableName is db-qualified.
+ val srcName = srcTable.stripPrefix("default.")
+ val dstName = dstTable.stripPrefix("default.")
+ // An empty destination timeline falls back to the sentinel latest
commit "0", the realistic
+ // "target not yet synced" shape. The sentinel is never greater than
the source's latest, so
+ // the else branch runs, and findInstantsAfter("0") makes the whole
source timeline the
+ // catch-up set: both inserts are counted.
+ assertResult(s"Count difference now is count($srcName) -
count($dstName) == 0. Catach up count is 2")(
+ result.head.getString(0))
+ }
+ }
+ }
+
+ test("Test Call sync_validate when the target table is mor (known
limitation)") {
+ withTempDir { tmp =>
+ val srcTable = generateTableName
+ val dstTable = generateTableName
+ createTable(srcTable, s"${tmp.getCanonicalPath}/$srcTable")
+ spark.sql(s"insert into $srcTable select 1, 'a1', 10, 1000")
+
+ // Writes to a mor table land as deltacommits, so both of these are in
the catch-up range.
+ createTable(dstTable, s"${tmp.getCanonicalPath}/$dstTable", "mor")
+ spark.sql(s"insert into $dstTable select 1, 'a1', 10, 1000")
+ spark.sql(s"insert into $dstTable select 2, 'a2', 20, 2000")
+
+ // Known limitation: the catch-up counting hardcodes the commit action
when it rebuilds the
+ // instants, so it cannot resolve deltacommit instants and the procedure
fails on any mor
+ // target with catch-up commits. See #19635. A fix flips this test to
assert a count.
+ val e = intercept[HoodieIOException] {
+ spark.sql(
+ s"""call sync_validate(src_table => '$srcTable', dst_table =>
'$dstTable',
+ | mode => 'noop', hive_server_url => 'jdbc:hive2://unused',
hive_pass => 'x')"""
+ .stripMargin).collect()
+ }
+ // The instant timestamp varies per run, but the synthesized commit
action is the whole bug.
+ assert(e.getMessage.startsWith("Cannot find the instant["), e.getMessage)
+ assert(e.getMessage.endsWith("__commit__COMPLETED]"), e.getMessage)
+ }
+ }
+
+ test("Test Call sync_validate record-count modes fail fast on an unreachable
HiveServer2") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ createTable(tableName, s"${tmp.getCanonicalPath}/$tableName")
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+
+ // The mode dispatch counts records before the timelines are compared,
so a single table as
+ // both source and target is enough here: neither call survives the
record counting.
+ // Nothing reaches the network either: the url carries no port, so the
authority resolves to
+ // unused:-1 and the driver's transport setup rejects that port before
any connect.
+
+ // mode = 'complete' routes to the countRecords overload that declares
its connection as
+ // `var conn: Connection = null` and closes it in an unguarded `finally
{ conn.close() }`.
+ // When DriverManager.getConnection throws, conn is still null, so the
finally block raises a
+ // NullPointerException that replaces the real SQLException. This pins
that masking bug, see
+ // #19635. A fix flips this test to assert an SQLException is present in
the chain.
+ val completeFailure = intercept[Throwable] {
+ spark.sql(
+ s"""call sync_validate(src_table => '$tableName', dst_table =>
'$tableName',
+ | mode => 'complete', hive_server_url => 'jdbc:hive2://unused',
hive_pass => 'x')"""
+ .stripMargin).collect()
+ }
+ // Assert over the cause chain, never the top-level type: on Spark 3.4+
QueryExecution wraps a
+ // NullPointerException thrown by an eagerly executed command into a
SparkException
+ // [INTERNAL_ERROR]. Message text is off limits too, the NPE message is
null on JDK 11 and the
+ // Hive URI wording moves with hive.version.
+
assert(causeChain(completeFailure).exists(_.isInstanceOf[NullPointerException]),
+ chainTypes(completeFailure))
+ assert(!causeChain(completeFailure).exists(_.isInstanceOf[SQLException]),
+ chainTypes(completeFailure))
+
+ // mode = 'latestPartitions' routes to the sibling overload, which
obtains the connection
+ // before entering its try and closes it under an `if (conn != null)`
guard, so the connection
+ // failure surfaces as the SQLException it is instead of being masked.
+ val latestPartitionsFailure = intercept[Throwable] {
+ spark.sql(
+ s"""call sync_validate(src_table => '$tableName', dst_table =>
'$tableName',
+ | mode => 'latestPartitions', hive_server_url =>
'jdbc:hive2://unused', hive_pass => 'x')"""
+ .stripMargin).collect()
+ }
+
assert(causeChain(latestPartitionsFailure).exists(_.isInstanceOf[SQLException]),
+ chainTypes(latestPartitionsFailure))
+ }
+ }
+}
