Repository navigation
Spark 4.1: Add Merge WriteSummary to Snapshot Summary - #15014
Conversation
singhpk234
left a comment
singhpk234
left a comment
There was a problem hiding this comment.
This would be a nice info to have !
should we namespace theme like
spark.merge-into.num-target-rows-matched-rows-updated
| setIfPositive(operation, "numTargetRowsCopied", mergeSummary.numTargetRowsCopied()); | ||
| setIfPositive(operation, "numTargetRowsDeleted", mergeSummary.numTargetRowsDeleted()); | ||
| setIfPositive(operation, "numTargetRowsUpdated", mergeSummary.numTargetRowsUpdated()); | ||
| setIfPositive(operation, "numTargetRowsInserted", mergeSummary.numTargetRowsInserted()); | ||
| setIfPositive( | ||
| operation, "numTargetRowsMatchedUpdated", mergeSummary.numTargetRowsMatchedUpdated()); | ||
| setIfPositive( | ||
| operation, "numTargetRowsMatchedDeleted", mergeSummary.numTargetRowsMatchedDeleted()); | ||
| setIfPositive( | ||
| operation, | ||
| "numTargetRowsNotMatchedBySourceUpdated", | ||
| mergeSummary.numTargetRowsNotMatchedBySourceUpdated()); | ||
| setIfPositive( | ||
| operation, | ||
| "numTargetRowsNotMatchedBySourceDeleted", | ||
| mergeSummary.numTargetRowsNotMatchedBySourceDeleted()); |
There was a problem hiding this comment.
do we wanna document it here : https://iceberg.apache.org/spec/?h=snapshot+summary#optional-snapshot-summary-fields
camel vs snake case
There was a problem hiding this comment.
changed to snake case (its actually kebab case?)
Is it ok to document in another pr?
There was a problem hiding this comment.
Is it ok to document in another pr?
ofc :) !, thank you for changing it to kebab case.
yea, this works for me. looks like there is already spark.xxx for other things. thanks for the feedback, welcome other thoughts as well ! |
singhpk234
left a comment
singhpk234
left a comment
There was a problem hiding this comment.
mostly LGTM, thanks @szehon-ho !
I have a minor suggestion please let me WDYT ?
| private void setMergeSummaryProperties(SnapshotUpdate<?> operation, MergeSummary mergeSummary) { | ||
| setIfPositive( | ||
| operation, "spark.merge-into.num-target-rows-copied", mergeSummary.numTargetRowsCopied()); | ||
| setIfPositive( | ||
| operation, | ||
| "spark.merge-into.num-target-rows-deleted", | ||
| mergeSummary.numTargetRowsDeleted()); | ||
| setIfPositive( | ||
| operation, | ||
| "spark.merge-into.num-target-rows-updated", | ||
| mergeSummary.numTargetRowsUpdated()); | ||
| setIfPositive( | ||
| operation, | ||
| "spark.merge-into.num-target-rows-inserted", | ||
| mergeSummary.numTargetRowsInserted()); | ||
| setIfPositive( | ||
| operation, | ||
| "spark.merge-into.num-target-rows-matched-updated", | ||
| mergeSummary.numTargetRowsMatchedUpdated()); | ||
| setIfPositive( | ||
| operation, | ||
| "spark.merge-into.num-target-rows-matched-deleted", | ||
| mergeSummary.numTargetRowsMatchedDeleted()); | ||
| setIfPositive( | ||
| operation, | ||
| "spark.merge-into.num-target-rows-not-matched-by-source-updated", | ||
| mergeSummary.numTargetRowsNotMatchedBySourceUpdated()); | ||
| setIfPositive( | ||
| operation, | ||
| "spark.merge-into.num-target-rows-not-matched-by-source-deleted", | ||
| mergeSummary.numTargetRowsNotMatchedBySourceDeleted()); | ||
| } | ||
|
|
||
| private void setIfPositive(SnapshotUpdate<?> operation, String key, long value) { | ||
| if (value >= 0) { | ||
| operation.set(key, String.valueOf(value)); | ||
| } | ||
| } |
There was a problem hiding this comment.
minor : I wonder if we should just move this to a util to have reuse and incase we add new metrics we just add in one place, thoughts ?
There was a problem hiding this comment.
yea good call, i made a new base class and put the duplicate method there
|
@singhpk234 can you take another look when you get a chance? thanks |
| import org.apache.spark.sql.connector.write.MergeSummary; | ||
|
|
||
| /** Base class for Spark write implementations with shared utility methods. */ | ||
| abstract class BaseSparkWrite { |
There was a problem hiding this comment.
Thank you so much for creating this base class !
singhpk234
left a comment
singhpk234
left a comment
There was a problem hiding this comment.
Looks good to me, thank you @szehon-ho !
|
Thanks @huaxingao @singhpk234 for reviews! |
3 participants
Spark 4.1 added "WriteSummary" API for certain DML operations, see https://issues.apache.org/jira/browse/SPARK-54309. Merge Into is the main one so far, hopefully more will follow.
This adds these operation metadata to Iceberg Snapshot Summary. It can be useful to debug data corruption issues, etc.
It is equivalent to the metadata stored in Delta Lake for debugging: https://docs.delta.io/delta-utility/#retrieve-delta-table-history
Welcome suggestions how to organize this summary