hudi-agent commented on code in PR #19161:
URL: https://github.com/apache/hudi/pull/19161#discussion_r3523711078
##########
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(
+ 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(row.size() == 1 && row.get(0).get(0) == 0)
+ val incCount = spark.sql(s"select count(1) from
$incViewName").collectAsList()
+ assert(incCount.size() == 1 && incCount.get(0).get(0) == 1)
Review Comment:
🤖 The test result itself supports this reading: if the range filtered on
instant time, `(000, commits.last]` would cover both commits' instant times and
return 2 records — but `incCount == 1`, which only makes sense if filtering is
on completion time and the last commit's completion time falls above `end` (its
own instant time). That points to the surviving record being the first commit
(id=1), so asserting the actual id here would be worth doing to nail down
whether the comment's "only the last commit" wording is inverted.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.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.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, s: StructType): Seq[Row] =
Review Comment:
🤖 nit: the parameter `s` in `keep` is a bit terse — could you rename it to
`schema` so the helper signature is self-documenting? Several test bodies also
use `s` as a local variable name, which makes call-sites like `keep(rows,
"...", s)` slightly harder to parse at a glance.
<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"` —
if this mirrors a literal string in the production code, could you fix the typo
there too (and update both assertions on lines 70 and 91)?
<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]
