hudi-agent commented on code in PR #19161:
URL: https://github.com/apache/hudi/pull/19161#discussion_r3533066932
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCopyToTempViewProcedure.scala:
##########
@@ -177,4 +177,116 @@ class TestCopyToTempViewProcedure extends
HoodieSparkSqlTestBase {
}
}
+
+ test("Test Call copy_to_temp_view Procedure with incremental query type") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | location '${tmp.getCanonicalPath}/$tableName'
+ | tblproperties (
+ | primaryKey = 'id',
+ | preCombineField = '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")
+
+ // Hudi incremental read treats the begin instant as an exclusive lower
bound, so a range from
+ // before the first commit up to the last commit returns only the
record(s) written by the last
+ // commit here (the first commit falls on / below the begin boundary and
is not re-emitted).
+ val row = spark.sql(
Review Comment:
🤖 nit: `collectAsList()` returns a `java.util.List[Row]`, so naming the
variable `row` (singular) is a bit misleading — could you rename it to `rows`
or `result`? The same pattern repeats in the `read_optimized` and
`as_of_instant` tests further down.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ExportInstantsProcedure.scala:
##########
@@ -88,19 +88,17 @@ class ExportInstantsProcedure extends BaseProcedure with
ProcedureBuilder with L
if (!new File(localFolder).isDirectory) throw new
HoodieException(localFolder + " is not a valid local directory")
- // The non archived instants can be listed from the Timeline. Wrap in a
mutable list because the
- // desc branch below reverses it in place via Collections.reverse, which
fails on the read-only
- // list produced by asJava over an immutable Scala collection.
- val nonArchivedInstants: util.List[HoodieInstant] = new
util.ArrayList[HoodieInstant](metaClient
+ // The non archived instants can be listed from the Timeline.
+ val nonArchivedInstants: util.List[HoodieInstant] = metaClient
.getActiveTimeline
.filterCompletedInstants.getInstants.iterator().asScala
.filter((i: HoodieInstant) => actionSet.contains(i.getAction))
- .toList.asJava)
+ .toList.asJava
Review Comment:
🤖 Dropping the `new util.ArrayList(...)` wrapper makes `nonArchivedInstants`
(and `archivedStatuses` on line 100) a read-only `asJava` view over an
immutable Scala list, so `Collections.reverse(...)` at lines 104/107 calls
`set()` and throws `UnsupportedOperationException` — `export_instants(..., desc
=> true)` now fails at runtime, which is exactly what the removed comment
warned about. This PR also removes the desc-ordering test that covered it, so
nothing catches the regression. Was this revert intentional?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestValidateHoodieSyncProcedure.scala:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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
+
+/**
+ * Tests for
[[org.apache.spark.sql.hudi.command.procedures.ValidateHoodieSyncProcedure]].
+ *
+ * The "complete" / "latestPartitions" modes require a live Hive/JDBC
endpoint, which is not
+ * available in the unit-test environment. Passing any other mode
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 {
+
+ private def createTable(tableName: String, path: String): Unit = {
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | location '$path'
+ | tblproperties (
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ | )
+ """.stripMargin)
+ }
+
+ test("Test Call sync_validate when the target table is ahead (catch-up
commits)") {
+ withTempDir { tmp =>
+ 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)
+ val message = result.head.getString(0)
+ assert(message.startsWith("Count difference now is"),
+ s"Unexpected message: $message")
+ // The destination has commits after the source's latest, so a catch-up
count is reported.
+ assert(message.contains("Catach up count is"), s"Unexpected message:
$message")
Review Comment:
🤖 nit: `"Catach up count is"` looks like a typo for `"Catch up count is"` —
could you double-check whether the production message itself has this
misspelling? If so, it might be worth fixing both together; if not, the
assertion string needs correcting.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestExportInstantsProcedure.scala:
##########
@@ -17,116 +17,34 @@
package org.apache.spark.sql.hudi.procedure
-import org.apache.spark.sql.Row
-
-import java.io.File
-
class TestExportInstantsProcedure extends HoodieSparkProcedureTestBase {
- private def createCowTable(tableName: String, path: String): Unit = {
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- |) using hudi
- | location '$path'
- | tblproperties (
- | primaryKey = 'id',
- | orderingFields = 'ts'
- | )
- """.stripMargin)
- }
-
- private def newExportDir(tmp: File, name: String): File = {
- val dir = new File(tmp, name)
- assert(dir.mkdirs(), s"Failed to create export dir $dir")
- dir
- }
-
- private def exportedCount(result: Array[Row]): Int = {
- assertResult(1)(result.length)
- val detail = result.head.getString(0)
- val matched = "Exported (\\d+) Instants".r.findFirstMatchIn(detail)
- assert(matched.isDefined, s"Unexpected export detail: $detail")
- matched.get.group(1).toInt
- }
-
test("Test Call export_instants Procedure") {
withTempDir { tmp =>
Review Comment:
🤖 nit: `// create table` and `// insert data to table` (a few lines below)
just restate what the code immediately says — could you drop them? The SQL and
insert statements are self-documenting here.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
