hudi-agent commented on code in PR #20279:
URL: https://github.com/apache/hudi/pull/20279#discussion_r4237328070
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ValidateHoodieSyncProcedure.scala:
##########
@@ -120,83 +119,80 @@ class ValidateHoodieSyncProcedure extends BaseProcedure
with ProcedureBuilder wi
if (commitsToCatchup.isEmpty) {
s"Count difference now is count(${target.getTableConfig.getTableName}) -
count(${source.getTableConfig.getTableName}) == ${targetCount - sourceCount}"
} else {
- val newInserts = countNewRecords(target, commitsToCatchup.map(elem =>
elem.requestedTime))
+ val newInserts = countNewRecords(target, commitsToCatchup)
s"Count difference now is count(${target.getTableConfig.getTableName}) -
count(${source.getTableConfig.getTableName}) == ${targetCount - sourceCount}" +
s". Catach up count is $newInserts"
}
}
+ @throws[SQLException]
def countRecords(jdbcUrl: String, source: HoodieTableMetaClient, dbName:
String, user: String, pass: String): Long = {
- var conn: Connection = null
- var rs: ResultSet = null
- var count: Long = -1
- try {
- conn = DriverManager.getConnection(jdbcUrl, user, pass)
- val stmt = conn.createStatement()
-
- stmt.execute("set
hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat");
- stmt.execute("set hive.stats.autogather=false");
-
- rs = stmt.executeQuery(
- s"select count(`_hoodie_commit_time`) as cnt from
$dbName.${source.getTableConfig.getTableName}")
- if (rs.next()) {
- count = rs.getLong("cnt")
- }
-
- println(s"Total records in ${source.getTableConfig.getTableName} is
$count")
- } finally {
- conn.close()
- if (rs != null) {
- rs.close()
- }
- }
+ val count = executeCountQuery(jdbcUrl, user, pass,
+ s"select count(`_hoodie_commit_time`) as cnt from
$dbName.${source.getTableConfig.getTableName}")
+ println(s"Total records in ${source.getTableConfig.getTableName} is
$count")
count
}
@throws[SQLException]
def countRecords(jdbcUrl: String, source: HoodieTableMetaClient, srcDb:
String, partitions: Int, user: String, pass: String): Long = {
- def getDate(dateTime: DateTime): String = {
-
s"${dateTime.getYear}-${dateTime.getMonthOfYear}%02d-${dateTime.getDayOfMonth}%02d"
- }
var dateTime = DateTime.now
- val endDateStr = getDate(dateTime)
+ val endDateStr = ValidateHoodieSyncProcedure.formatDatestr(dateTime)
dateTime = dateTime.minusDays(partitions)
- val startDateStr = getDate(dateTime)
+ val startDateStr = ValidateHoodieSyncProcedure.formatDatestr(dateTime)
println("Start date " + startDateStr + " and end date " + endDateStr)
countRecords(jdbcUrl, source, srcDb, startDateStr, endDateStr, user, pass)
}
@throws[SQLException]
private def countRecords(jdbcUrl: String, source: HoodieTableMetaClient,
srcDb: String, startDateStr: String, endDateStr: String, user: String, pass:
String): Long = {
- var rs: ResultSet = null
+ executeCountQuery(jdbcUrl, user, pass,
+ s"select count(`_hoodie_commit_time`) as cnt from
$srcDb.${source.getTableConfig.getTableName}" +
+ s" where datestr>'$startDateStr' and datestr<='$endDateStr'")
+ }
+
+ /**
+ * Runs a `count(...) as cnt` query over a new HiveServer2 connection and
returns `cnt`, or -1 if
+ * the query yields no row.
+ *
+ * Each JDBC resource is closed only once it has been opened, in the reverse
order of opening, so a
+ * failure to connect surfaces as the [[SQLException]] it is rather than as
a failure to close a
+ * connection that was never opened.
+ */
+ @throws[SQLException]
+ private def executeCountQuery(jdbcUrl: String, user: String, pass: String,
countQuery: String): Long = {
+ val conn: Connection = DriverManager.getConnection(jdbcUrl, user, pass)
try {
- val conn = DriverManager.getConnection(jdbcUrl, user, pass)
- val stmt = conn.createStatement
- try { // stmt.execute("set mapred.job.queue.name=<queue_name>");
+ val stmt: Statement = conn.createStatement()
+ try {
stmt.execute("set
hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat")
stmt.execute("set hive.stats.autogather=false")
- rs = stmt.executeQuery(s"select count(`_hoodie_commit_time`) as cnt
from $srcDb.${source.getTableConfig.getTableName} where datestr>'$startDateStr'
and datestr<='$endDateStr'")
- if (rs.next)
- rs.getLong("cnt")
- else
- -1
+ val rs: ResultSet = stmt.executeQuery(countQuery)
+ try {
+ if (rs.next()) rs.getLong("cnt") else -1
+ } finally {
+ rs.close()
+ }
} finally {
- if (rs != null) rs.close()
- if (conn != null) conn.close()
- if (stmt != null) stmt.close()
+ stmt.close()
}
+ } finally {
+ conn.close()
}
}
+ /**
+ * Sums the records inserted by the given write instants. Each instant is
read as it is on the
+ * timeline, so the deltacommits of a merge-on-read table and the
replacecommits of a clustering
+ * or an insert overwrite are resolved under their own action. Pending
instants are skipped, as
+ * they have not committed any records yet.
+ */
@throws[IOException]
- def countNewRecords(target: HoodieTableMetaClient, commitsToCatchup:
List[String]): Long = {
+ def countNewRecords(target: HoodieTableMetaClient, commitsToCatchup:
List[HoodieInstant]): Long = {
var totalNew: Long = 0
- val timeline: HoodieTimeline =
target.reloadActiveTimeline.getCommitAndReplaceTimeline.filterCompletedInstants
- for (commit <- commitsToCatchup) {
- val instantGenerator = target.getTimelineLayout.getInstantGenerator
- val instant: HoodieInstant =
instantGenerator.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, commit)
- val c: HoodieCommitMetadata = timeline.readCommitMetadata(instant)
+ val timeline: HoodieTimeline =
target.reloadActiveTimeline.getCommitsTimeline.filterCompletedInstants
Review Comment:
🤖 nit: could you rename `c` to `metadata`? It's the commit metadata being
summed, and a single letter is easy to misread here.
<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/ValidateHoodieSyncProcedure.scala:
##########
@@ -120,83 +119,80 @@ class ValidateHoodieSyncProcedure extends BaseProcedure
with ProcedureBuilder wi
if (commitsToCatchup.isEmpty) {
s"Count difference now is count(${target.getTableConfig.getTableName}) -
count(${source.getTableConfig.getTableName}) == ${targetCount - sourceCount}"
} else {
- val newInserts = countNewRecords(target, commitsToCatchup.map(elem =>
elem.requestedTime))
+ val newInserts = countNewRecords(target, commitsToCatchup)
s"Count difference now is count(${target.getTableConfig.getTableName}) -
count(${source.getTableConfig.getTableName}) == ${targetCount - sourceCount}" +
s". Catach up count is $newInserts"
}
}
+ @throws[SQLException]
def countRecords(jdbcUrl: String, source: HoodieTableMetaClient, dbName:
String, user: String, pass: String): Long = {
- var conn: Connection = null
- var rs: ResultSet = null
- var count: Long = -1
- try {
- conn = DriverManager.getConnection(jdbcUrl, user, pass)
- val stmt = conn.createStatement()
-
- stmt.execute("set
hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat");
- stmt.execute("set hive.stats.autogather=false");
-
- rs = stmt.executeQuery(
- s"select count(`_hoodie_commit_time`) as cnt from
$dbName.${source.getTableConfig.getTableName}")
- if (rs.next()) {
- count = rs.getLong("cnt")
- }
-
- println(s"Total records in ${source.getTableConfig.getTableName} is
$count")
- } finally {
- conn.close()
- if (rs != null) {
- rs.close()
- }
- }
+ val count = executeCountQuery(jdbcUrl, user, pass,
+ s"select count(`_hoodie_commit_time`) as cnt from
$dbName.${source.getTableConfig.getTableName}")
+ println(s"Total records in ${source.getTableConfig.getTableName} is
$count")
count
}
@throws[SQLException]
def countRecords(jdbcUrl: String, source: HoodieTableMetaClient, srcDb:
String, partitions: Int, user: String, pass: String): Long = {
- def getDate(dateTime: DateTime): String = {
-
s"${dateTime.getYear}-${dateTime.getMonthOfYear}%02d-${dateTime.getDayOfMonth}%02d"
- }
var dateTime = DateTime.now
- val endDateStr = getDate(dateTime)
+ val endDateStr = ValidateHoodieSyncProcedure.formatDatestr(dateTime)
dateTime = dateTime.minusDays(partitions)
- val startDateStr = getDate(dateTime)
+ val startDateStr = ValidateHoodieSyncProcedure.formatDatestr(dateTime)
println("Start date " + startDateStr + " and end date " + endDateStr)
countRecords(jdbcUrl, source, srcDb, startDateStr, endDateStr, user, pass)
}
@throws[SQLException]
private def countRecords(jdbcUrl: String, source: HoodieTableMetaClient,
srcDb: String, startDateStr: String, endDateStr: String, user: String, pass:
String): Long = {
- var rs: ResultSet = null
+ executeCountQuery(jdbcUrl, user, pass,
+ s"select count(`_hoodie_commit_time`) as cnt from
$srcDb.${source.getTableConfig.getTableName}" +
+ s" where datestr>'$startDateStr' and datestr<='$endDateStr'")
+ }
+
+ /**
+ * Runs a `count(...) as cnt` query over a new HiveServer2 connection and
returns `cnt`, or -1 if
+ * the query yields no row.
+ *
+ * Each JDBC resource is closed only once it has been opened, in the reverse
order of opening, so a
+ * failure to connect surfaces as the [[SQLException]] it is rather than as
a failure to close a
+ * connection that was never opened.
+ */
+ @throws[SQLException]
+ private def executeCountQuery(jdbcUrl: String, user: String, pass: String,
countQuery: String): Long = {
+ val conn: Connection = DriverManager.getConnection(jdbcUrl, user, pass)
try {
- val conn = DriverManager.getConnection(jdbcUrl, user, pass)
- val stmt = conn.createStatement
- try { // stmt.execute("set mapred.job.queue.name=<queue_name>");
+ val stmt: Statement = conn.createStatement()
+ try {
stmt.execute("set
hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat")
stmt.execute("set hive.stats.autogather=false")
- rs = stmt.executeQuery(s"select count(`_hoodie_commit_time`) as cnt
from $srcDb.${source.getTableConfig.getTableName} where datestr>'$startDateStr'
and datestr<='$endDateStr'")
- if (rs.next)
- rs.getLong("cnt")
- else
- -1
+ val rs: ResultSet = stmt.executeQuery(countQuery)
+ try {
+ if (rs.next()) rs.getLong("cnt") else -1
+ } finally {
+ rs.close()
+ }
} finally {
- if (rs != null) rs.close()
- if (conn != null) conn.close()
- if (stmt != null) stmt.close()
+ stmt.close()
}
+ } finally {
+ conn.close()
}
}
+ /**
+ * Sums the records inserted by the given write instants. Each instant is
read as it is on the
+ * timeline, so the deltacommits of a merge-on-read table and the
replacecommits of a clustering
+ * or an insert overwrite are resolved under their own action. Pending
instants are skipped, as
+ * they have not committed any records yet.
+ */
@throws[IOException]
- def countNewRecords(target: HoodieTableMetaClient, commitsToCatchup:
List[String]): Long = {
+ def countNewRecords(target: HoodieTableMetaClient, commitsToCatchup:
List[HoodieInstant]): Long = {
var totalNew: Long = 0
- val timeline: HoodieTimeline =
target.reloadActiveTimeline.getCommitAndReplaceTimeline.filterCompletedInstants
- for (commit <- commitsToCatchup) {
- val instantGenerator = target.getTimelineLayout.getInstantGenerator
- val instant: HoodieInstant =
instantGenerator.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, commit)
- val c: HoodieCommitMetadata = timeline.readCommitMetadata(instant)
+ val timeline: HoodieTimeline =
target.reloadActiveTimeline.getCommitsTimeline.filterCompletedInstants
+ for (instant <- commitsToCatchup if instant.isCompleted) {
+ val c: HoodieCommitMetadata = TimelineUtils.getCommitMetadata(instant,
timeline)
totalNew += c.fetchTotalRecordsWritten - c.fetchTotalUpdateRecordsWritten
Review Comment:
🤖 non-blocking: now that clustering replacecommits and MOR compaction
commits resolve instead of throwing, `fetchTotalRecordsWritten -
fetchTotalUpdateRecordsWritten` counts every record they rewrite as a new
insert, so a catch-up range with a clustering or compaction would overstate the
count; would it make sense to skip instants whose `getOperationType` is
`CLUSTER`, `COMPACT` or `LOG_COMPACT`?
<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]
