Skip to content

[SPARK-40826][SS] Add additional checkpoint rename file check #38291

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
viirya wants to merge 5 commits into from
Closed

[SPARK-40826][SS] Add additional checkpoint rename file check #38291

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
fa5b387
Add additional renamed file check
viirya Oct 18, 2022
7535fa9
Change config name
viirya Oct 18, 2022
9028766
Add version
viirya Oct 18, 2022
115c0e0
Trigger Build
viirya Oct 18, 2022
a2a7ed2
Merge remote-tracking branch 'upstream/master' into add_file_check
viirya Oct 19, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1830,6 +1830,14 @@ object SQLConf {
.stringConf
.createWithDefault("lz4")

val CHECKPOINT_RENAMEDFILE_CHECK_ENABLED =
buildConf("spark.sql.streaming.checkpoint.renamedFileCheck.enabled")
.doc("When true, Spark will validate if renamed checkpoint file exists.")
.internal()
.version("3.4.0")
.booleanConf
.createWithDefault(false)

/**
* Note: this is defined in `RocksDBConf.FORMAT_VERSION`. These two places should be updated
* together.
Expand Down Expand Up @@ -4234,6 +4242,8 @@ class SQLConf extends Serializable with Logging {

def stateStoreCompressionCodec: String = getConf(STATE_STORE_COMPRESSION_CODEC)

def checkpointRenamedFileCheck: Boolean = getConf(CHECKPOINT_RENAMEDFILE_CHECK_ENABLED)

def parquetFilterPushDown: Boolean = getConf(PARQUET_FILTER_PUSHDOWN_ENABLED)

def parquetFilterPushDownDate: Boolean = getConf(PARQUET_FILTER_PUSHDOWN_DATE_ENABLED)
Expand Down
7 changes: 7 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CheckpointFileManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,13 @@ object CheckpointFileManager extends Logging {
s"Failed to rename temp file $tempPath to $finalPath because file exists", fe)
if (!overwriteIfPossible) throw fe
}

// Optionally, check if the renamed file exists
if (SQLConf.get.checkpointRenamedFileCheck && !fm.exists(finalPath)) {
throw new IllegalStateException(s"Renamed temp file $tempPath to $finalPath. " +
s"But $finalPath does not exist.")
}

logInfo(s"Renamed temp file $tempPath to $finalPath")
} finally {
terminated = true
Expand Down