hudi-agent commented on code in PR #19161:
URL: https://github.com/apache/hudi/pull/19161#discussion_r3522667895
##########
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:
🤖 I think this explanation is inverted. `begin => '000'` sits below every
completion time, so the open lower bound drops nothing — it's the `end`
boundary that excludes a commit here: `end_instance_time` is `commits.last` (an
*instant* time), but the incremental range filters on *completion* time, and
the last commit's completion time is greater than its own instant time, so it's
the *last* commit that gets dropped and the surviving row is the *first*
commit's (id=1). Could you assert the returned id to confirm? If it's 1, the
"only the last commit" wording is backwards.
<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]
