[SPARK-58770][CORE] Assign a name to the error condition _LEGACY_ERROR_TEMP_3016-3020

[LuciferYang]

What changes were proposed in this pull request?

This PR converts the five _LEGACY_ERROR_TEMP_* conditions in SparkCoreErrors that cover RDD checkpointing, continuing the cleanup under SPARK-37935. Four get user-facing names; the fifth is an unreachable defensive branch and becomes an internal error.

Legacy	Builder	Now	SQLSTATE
_LEGACY_ERROR_TEMP_3016	checkpointDirectoryHasNotBeenSetInSparkContextError	CHECKPOINT_DIRECTORY_NOT_SET	55019
_3017	invalidCheckpointDirectoryError	INVALID_CHECKPOINT_DIRECTORY	58030
_3018	failToCreateCheckpointPathError	FAILED_CREATE_CHECKPOINT_DIRECTORY	58030
_3019	checkpointRDDHasDifferentNumberOfPartitionsFromOriginalRDDError	CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH	58030
_3020	mustSpecifyCheckpointDirError	INTERNAL_ERROR (entry deleted)	XX000

Four top-level names rather than one umbrella: sqlState lives on the umbrella, so grouping them would force a single SQLSTATE and lose the 55019/58030 split, and no umbrella sentence holds for all four. _3016 fires before any job runs and is a user configuration mistake; the other three are storage failures during an action.

invalidCheckpointDirectoryError gains an expectedFileName parameter, taken from the checkpointFileName(i) the caller already computes, and reports the directory and file name separately. The old message named only the offending path, which is misleading here: the check walks the sorted part-* files and compares the i-th name against part-%05d(i), so the path it reports is a perfectly valid file sitting where a missing one should be. With part-00001 deleted from a 4-partition checkpoint the old message read Invalid checkpoint file: .../part-00002 and sent the reader after the wrong file. The builder is renamed alongside the condition because its old name contradicted it; the other three keep their Scala names, which already agree with theirs. All four have a single call site, so the diff size does not distinguish them.

failToCreateCheckpointPathError's parameter is renamed checkpointDirPath to path, matching INVALID_BUCKET_FILE and the other path-carrying conditions.

Reachability, per condition

• _3016, user configuration.RDD.checkpoint() is public, and sc.checkpointDir is set through SparkContext.setCheckpointDir, reachable either by calling it directly or, since 4.0.0, through spark.checkpoint.dir, which SparkContext.scala:623 forwards to the same setter. Calling checkpoint() with neither lands here. Reachable from Dataset.checkpoint too, which calls internalRdd.checkpoint(), so the message deliberately avoids naming RDDs. The old text mentioned only setCheckpointDir; the new one names the conf as well, which matters for a Connect client that cannot call the setter.
• _3017, incomplete checkpoint directory.getPartitions requires the part-* files to be a contiguous part-00000..part-000NN. It fires on a partially written checkpoint, a manually pruned directory, or a directory handed to SparkContext.checkpointFile (which is how streaming recovery rebuilds generatedRDDs). Driver-side: getPartitions runs from RDD.partitions, and partitions_ is @transient, so executors never compute it.
• _3018, storage refused the directory. Only fires where a FileSystem reports failure by returning false from mkdirs rather than throwing, which is what HDFS and S3A can do; LocalFileSystem tends to throw instead. Driver-side, inside RDD.doCheckpoint() at the end of the first action.
• _3019, the write and the read-back disagree. Not an engine invariant. The driver creates the directory, each executor writes its own part-* through its own FileSystem, and then the driver counts what its FileSystem lists, so the two sides of the comparison resolve in different JVMs. setCheckpointDir only warns when a cluster-mode application points at a local path, so a user who sets /tmp/ckpt on a cluster reaches this directly: the executors write to their own disks and the driver lists an empty directory. getPartitions' own scaladoc states the assumption being verified ("assumes that the original set of checkpoint files are fully preserved in a reliable storage"). Converting this one to INTERNAL_ERROR would report a storage misconfiguration as a Spark bug.
• _3020, unreachable.ReliableRDDCheckpointData's cpDir field throws when sc.checkpointDir is None, but its only construction site is RDD.scala:1743, two lines below the context.checkpointDir.isEmpty guard that raises _3016, inside the same RDDCheckpointData.synchronized block; cpDir is a plain val, evaluated there. mustSpecifyCheckpointDirError has a single call site, the getOrElse on that field. It is reachable only by a cross-thread race that also makes the condition a duplicate of _3016, so it gets internalError rather than a second user-facing name for the same situation.

Why are the changes needed?

The error-conditions README disallows new _LEGACY_ERROR_TEMP_* entries and asks existing ones to be resolved. This clears five of them.

Three of the five were also weak on their own terms. _3016 predates spark.checkpoint.dir and told the user about only one of the two ways to configure a directory. _3017 pointed at the wrong file, as described above. _3018 said only that creating the path failed, without saying that the filesystem reported it through a return value, which is the detail that tells an operator to look at permissions rather than at Spark.

Does this PR introduce any user-facing change?

Yes, to error messages, with no API change.

Converting any legacy condition changes the rendered string in two mechanical ways: SparkThrowableHelper.formatErrorMessage suppresses the [CONDITION] prefix only for _LEGACY_ERROR_-prefixed names, and appends SQLSTATE: xxxxx when a sqlState exists (legacy entries have none, so all four gain both). Beyond that:

• _3016: Checkpoint directory has not been set in the SparkContext becomes Cannot checkpoint because no checkpoint directory is configured. Set one with SparkContext.setCheckpointDir or the "spark.checkpoint.dir" configuration.
• _3017: Invalid checkpoint file: <path> becomes Cannot read the checkpoint directory <path>: expected the partition file <expectedFileName> but found <fileName>. The partition files must be numbered contiguously from part-00000. The parameter set changes from one key to three, so getMessageParameters() gains expectedFileName and fileName while path narrows from the file to its directory.
• _3018: Failed to create checkpoint path <checkpointDirPath> becomes Failed to create the checkpoint directory <path> as FileSystem.mkdirs returned false. The parameter is renamed, so getMessageParameters() has path where it had checkpointDirPath.
• _3019: the three-line Checkpoint RDD has a different number of partitions from original RDD. Original RDD [ID: ..., num of partitions: ...]; Checkpoint RDD [ID: ..., num of partitions: ...]. becomes The checkpoint of RDD <originalRDDId> has <newRDDLength> partition(s), but the RDD itself has <originalRDDLength>. The checkpoint RDD is <newRDDId>. plus a second line naming the two usual causes. Same four parameters.
• _3020: Checkpoint dir must be specified. becomes [INTERNAL_ERROR] SparkContext.checkpointDir is unset when creating ReliableRDDCheckpointData. SQLSTATE: XX000, and getMessageParameters() goes from empty to {"message": ...} since INTERNAL_ERROR's template is <message>. This does not change any job-failure message shape: the throw happens on the driver inside RDD.checkpoint(), before any task runs, so DAGScheduler.abortStage's isInternalError filter is not involved.

How was this patch tested?

None of the five had any test coverage, and CheckpointSuite contained no intercept at all. Three tests are added to CheckpointStorageSuite, each asserting the condition and the SQLSTATE, and each failing against the pre-change code on both fields (the legacy entries carry no sqlState):

• "checkpoint() without a checkpoint directory" builds a context with no checkpoint directory and asserts CHECKPOINT_DIRECTORY_NOT_SET.
• "reading a checkpoint directory with a missing partition file" checkpoints a 4-partition RDD, deletes part-00001, then reads the directory back through SparkContext.checkpointFile and asserts INVALID_CHECKPOINT_DIRECTORY with all three message parameters, pinning that the reported expectation is part-00001 and the file found is part-00002.
• "checkpoint path that cannot be created" registers a LocalFileSystem subclass whose mkdirs returns false for the per-RDD directory and asserts FAILED_CREATE_CHECKPOINT_DIRECTORY. Two details are load-bearing: LocalFileSystem throws FileAlreadyExistsException when the path is occupied instead of returning false, so the failure has to be injected rather than staged on disk; and Hadoop caches FileSystem instances per scheme, so the test also sets fs.file.impl.disable.cache=true or an earlier test's real LocalFileSystem is used instead.

CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH has no triggering test. The comparison it guards needs an original RDD to count against, and on the read-back path there is none: getPartitions' own scaladoc notes there is "no way to know a priori the number of partitions to expect". Reproducing the write-path case means making the driver and the executors see different contents for the same directory, which local mode cannot do since both sides share one filesystem. Two related gaps in the same check are filed separately, since both are pre-existing and neither is a naming change: SPARK-58881, a non-numeric part-* name throws a raw NumberFormatException from the sortBy before the validation loop runs; and SPARK-58883, deleting the trailing part-* file leaves the rest contiguous, so a read through SparkContext.checkpointFile silently yields fewer partitions. Detecting the latter needs the expected partition count persisted at write time, i.e. a new on-disk format.

The SQLSTATE assertions were confirmed to be live by temporarily setting CHECKPOINT_DIRECTORY_NOT_SET's value to 42000 and watching the test fail with sqlState: expected '55019' but got '42000' before restoring it. checkError skips the comparison when sqlState is None and SparkThrowableSuite only checks that a state is registered, so a wrong SQLSTATE would otherwise ship green.

Ran core/testOnly org.apache.spark.SparkThrowableSuite org.apache.spark.CheckpointSuite org.apache.spark.CheckpointStorageSuite (68 tests, all passing).

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 4.8)

[uros-b]

LGTM, thank you @LuciferYang!

[dongjoon-hyun]

I reviewed this PR. Overall it looks solid: the four JSON entries are inserted at the correct alphabetical positions, both SQLSTATEs (55019, 58030) are registered in error-states.json, every message placeholder matches the Scala messageParameters keys, no references to _LEGACY_ERROR_TEMP_3016..3020 remain anywhere in the repo, and no MiMa exclusion is needed since SparkCoreErrors is private[spark]. I also verified the new tests' path-string assertions hold (the listStatus-qualified parent matches the qualified checkpoint dir exactly, including on macOS /var vs /private/var).

A few findings, in decreasing order of importance:

1. getPartitions can throw a raw NumberFormatException before the new INVALID_CHECKPOINT_FILE fires (pre-existing, but surfaced by this change)

In ReliableCheckpointRDD.scala (line 78), a stray non-numeric part-* file, e.g. a leftover part-00000.bak, passes the startsWith("part-") filter, and sortBy(_.getName.stripPrefix("part-").toInt) throws java.lang.NumberFormatException before the validation loop this PR modifies can raise the new condition. The new message states "The partition files must be numbered contiguously from part-00000", which advertises validation the code cannot deliver for that input. Consider tightening the filter (e.g. to names matching part-\d+) so such files reach the new error path instead, either here or as a follow-up.

2. The new test re-implements the rdd-<id> path convention

In the "checkpoint path that cannot be created" test, new Path(sc.getCheckpointDir.get, s"rdd-${rdd.id}") hand-builds the per-RDD directory. ReliableRDDCheckpointData.checkpointPath(sc, rdd.id).get is accessible from the test's package, yields the identical string, and ties the assertion to the same code path production uses, so a future layout change cannot silently strand the expectation. (The startsWith("rdd-") check inside MkdirsFailingFilesystem cannot use the helper since Hadoop instantiates it reflectively, so that one hardcode is unavoidable.)

3. Nit: namespace of the new names

CHECKPOINT_DIRECTORY_NOT_SET, INVALID_CHECKPOINT_FILE, and FAILED_CREATE_CHECKPOINT_DIRECTORY claim generic checkpoint names for RDD-specific errors, while the streaming analog _LEGACY_ERROR_TEMP_1298 ("checkpointLocation must be specified ...") is still awaiting a name and condition names are frozen once shipped. Precedent cuts both ways (CHECKPOINT_RDD_BLOCK_ID_NOT_FOUND is RDD-side with a generic prefix; streaming uses STREAMING_CHECKPOINT_*), so this is only a naming-foresight consideration, and the streaming case would likely be named STREAMING_CHECKPOINT_LOCATION_NOT_SET anyway.

4. Nit: the internal error text keeps the legacy user-directive phrasing

SparkException.internalError("Checkpoint dir must be specified.") reads as an instruction to the user, but this branch only fires on a broken invariant. Other internalError sites in core describe the anomalous state ("memory store not initialized yet", "Index file is deleted already."), so something like "SparkContext.checkpointDir is unset when creating ReliableRDDCheckpointData" would give a bug report actual diagnostic content.

[dongjoon-hyun]

I re-reviewed at a1dd9bf. Findings #2 (the test hand-building the rdd-<id> path) and #4 (the internal error text) from my earlier review are addressed, and #3 was partly answered by the rename. Re-verified the mechanical parts against the new HEAD: the four JSON entries are still at the correct alphabetical positions (checked the whole key list, not just the neighbors), the file is still in the generator's format, 55019 and 58030 are both registered in error-states.json and match existing usage (58030 is already shared by INVALID_BUCKET_FILE, CANNOT_LOAD_STATE_STORE and 11 others), every placeholder matches a messageParameters key, no _LEGACY_ERROR_TEMP_3016..3020 reference and no occurrence of the five old message strings remains anywhere in the repo, docs/sql-error-conditions.md needs no update since it is generated by build-error-docs.py, and all 33 check runs on a1dd9bf are green.

Two things left, one of which blocks the merge.

1. Please refresh the PR description before merging — it still describes the pre-rename state

dev/merge_spark_pr.py uses the PR body as the commit message, so as written the commit log will name a condition that does not exist. Three spots:

• the table row: | _3017 | invalidCheckpointFileError | INVALID_CHECKPOINT_FILE | 58030 | — the builder is now invalidCheckpointDirectoryError and the condition is INVALID_CHECKPOINT_DIRECTORY
• "invalidCheckpointFileError gains an expectedFileName parameter"
• the testing section: "asserts INVALID_CHECKPOINT_FILE with all three message parameters"

2. My earlier finding #1 is still open, and the rename sharpened it

The new message ends with "The partition files must be numbered contiguously from part-00000, one per partition", but the check in ReliableCheckpointRDD.getPartitions (line 75-87) does not deliver that in two directions:

• Non-numeric part-* file. A stray part-00000.bak passes the startsWith("part-") filter, and sortBy(_.getName.stripPrefix("part-").toInt) throws a raw NumberFormatException before the validation loop runs, so the new condition never fires.
• Missing trailing file. Delete part-00003 from a 4-partition checkpoint and the remaining names are still contiguous, so the loop passes and Array.tabulate(inputFiles.length) quietly yields a 3-partition RDD. On the write path CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH catches that, but on the read-back path (SparkContext.checkpointFile, which is how streaming recovery rebuilds generatedRDDs) there is no original RDD to compare against and the data is silently truncated.

Both are pre-existing, but widening the name from INVALID_CHECKPOINT_FILE to INVALID_CHECKPOINT_DIRECTORY implies the whole directory is validated, which makes the gap more visible than before. Either tighten the filter to part-\d+ so the first case reaches the new error path (a follow-up is fine), or trim the last sentence to claim only the contiguity the loop actually checks.

3. Nit, on the untested condition

The argument for leaving CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH uncovered — that a fake FileSystem would test the fake — reads oddly next to the FAILED_CREATE_CHECKPOINT_DIRECTORY test, which injects the failure exactly the same way by making mkdirs return false. A LocalFileSystem subclass whose listStatus hides the last part-* would exercise the real comparison in writeRDDToCheckpointDirectory (line 178-182), not the fake. Not asking for the test, just for the reasoning in the description to line up with what the neighbouring test does.

Otherwise this looks good to me. Splitting _3017's parameters into (directory, expected name, found name) is a real fix rather than a rename — the old message pointed at a perfectly valid file — and demoting _3020 to internalError is right, given that the only construction site of ReliableRDDCheckpointData sits two lines below the context.checkpointDir.isEmpty guard in RDD.checkpoint().

[LuciferYang]

@dongjoon-hyun Fixed the description and trimmed the message. Both gaps in finding 2 are now filed.

1. Description

Updated the three spots you listed, plus four more I found while checking: SparkContext.scala:614 should have been :623, the _3016 bullet claimed sc.checkpointDir "has exactly two writers" when spark.checkpoint.dir is a second entry point to the same setter rather than a second writer, the _3020 bullet in the user-facing-change list was missing its old to new string, and the _3017 rendered string still carried the sentence I trimmed below.

I also dropped this sentence from the _3020 bullet: "The only writer that can store None is setCheckpointDir(null), which no production code calls, and the field is private[spark]." It is refutable with one grep, since checkpointDir is a private[spark] var and three mllib suites assign None to it directly. The reachability argument does not need it: one construction site, guarded by the isEmpty check in the same synchronized block, cpDir evaluated there, and one call site for the builder.

2. Both directions, filed separately

Trimmed the message to drop , one per partition, so it now claims only the contiguity the loop checks. That is the option you offered, and it is the honest one: the trailing-file case leaves the survivors "one per partition" while the RDD is short.

The two gaps are different problems, so they are separate tickets rather than one:

• SPARK-58881 tightens the filter to part- followed by digits. Not \d{5}: %05d is a minimum width, so more than 100000 partitions gives part-100000.
• SPARK-58883 covers the truncation you found. This one is worse than the first and cannot be fixed by a filter: detecting it needs the expected partition count persisted at write time, which is a new on-disk format with a compatibility story (a directory written by an earlier version has no such file). _partitioner is the precedent. Keeping that out of a naming PR.

3. Nit

You are right that the old wording did not line up. The reason it stays untested is not the fake filesystem; it is that the read-back path has no original RDD to count against, which getPartitions' scaladoc already states. Reworded, and the write-path case (driver and executors seeing different contents) is what local mode cannot reproduce.

[LuciferYang]

@dongjoon-hyun this one is ready to go, could you please take another look when you have time? Thanks

[dongjoon-hyun]

+1, LGTM

[LuciferYang]

Merge Summary:

• merged into master 5ef3a94
• merged into branch-4.x 5a88ff5

Posted by merge_spark_pr.py

[LuciferYang]

Thank you @dongjoon-hyun@uros-b