[SPARK-26824][SS]Fix the checkpoint location and _spark_metadata when it contains special chars #23733

zsxwing · 2019-02-04T17:58:04Z

What changes were proposed in this pull request?

When a user specifies a checkpoint location or a file sink output using a path containing special chars that need to be escaped in a path, the streaming query will store checkpoint and file sink metadata in a wrong place. In this PR, I uploaded a checkpoint that was generated by the following codes using Spark 2.4.0 to show this issue:

implicit val s = spark.sqlContext
val input = org.apache.spark.sql.execution.streaming.MemoryStream[Int]
input.addData(1, 2, 3)
val q = input.toDF.writeStream.format("parquet").option("checkpointLocation", ".../chk %@#chk").start(".../output %@#output")
q.stop()

Here is the structure of the directory:

sql/core/src/test/resources/structured-streaming/escaped-path-2.4.0
├── chk%252520%252525@%252523chk
│   ├── commits
│   │   └── 0
│   ├── metadata
│   └── offsets
│       └── 0
├── output %@#output
│   └── part-00000-97f675a2-bb82-4201-8245-05f3dae4c372-c000.snappy.parquet
└── output%20%25@%23output
    └── _spark_metadata
        └── 0

In this checkpoint, the user specified checkpoint location is .../chk %@#chk but the real path to store the checkpoint is .../chk%252520%252525@%252523chk (this is generated by escaping the original path three times). The user specified output path is .../output %@#output but the path to store _spark_metadata is .../output%20%25@%23output/_spark_metadata (this is generated by escaping the original path once). The data files are still in the correct path (such as .../output %@#output/part-00000-97f675a2-bb82-4201-8245-05f3dae4c372-c000.snappy.parquet).

This checkpoint will be used in unit tests in this PR.

The fix is just simply removing improper Path.toUri calls to fix the issue.

However, as the user may not read the release note and is not aware of this checkpoint location change, if they upgrade Spark without moving checkpoint to the new location, their query will just start from the scratch. In order to not surprise the users, this PR also adds a check to detect the impacted paths and throws an error to include the migration guide. This check can be turned off by an internal sql conf spark.sql.streaming.checkpoint.escapedPathCheck.enabled. Here are examples of errors that will be reported:

Streaming checkpoint error:

Error: we detected a possible problem with the location of your checkpoint and you
likely need to move it before restarting this query.

Earlier version of Spark incorrectly escaped paths when writing out checkpoints for
structured streaming. While this was corrected in Spark 3.0, it appears that your
query was started using an earlier version that incorrectly handled the checkpoint
path.

Correct Checkpoint Directory: /.../chk %@#chk
Incorrect Checkpoint Directory: /.../chk%252520%252525@%252523chk

Please move the data from the incorrect directory to the correct one, delete the
incorrect directory, and then restart this query. If you believe you are receiving
this message in error, you can disable it with the SQL conf
spark.sql.streaming.checkpoint.escapedPathCheck.enabled.

File sink error (_spark_metadata):

Error: we detected a possible problem with the location of your "_spark_metadata"
directory and you likely need to move it before restarting this query.

Earlier version of Spark incorrectly escaped paths when writing out the
"_spark_metadata" directory for structured streaming. While this was corrected in
Spark 3.0, it appears that your query was started using an earlier version that
incorrectly handled the "_spark_metadata" path.

Correct "_spark_metadata" Directory: /.../output %@#output/_spark_metadata
Incorrect "_spark_metadata" Directory: /.../output%20%25@%23output/_spark_metadata

Please move the data from the incorrect directory to the correct one, delete the
incorrect directory, and then restart this query. If you believe you are receiving
this message in error, you can disable it with the SQL conf
spark.sql.streaming.checkpoint.escapedPathCheck.enabled.

How was this patch tested?

The new unit tests.

zsxwing · 2019-02-04T18:00:47Z

cc @brkyvz

SparkQA · 2019-02-04T21:09:16Z

Test build #102038 has finished for PR 23733 at commit 3b357e5.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2019-02-04T22:06:16Z

Marking this WIP as I'm also working on improving errors to include the migration instructions for people who hit this issue.

SparkQA · 2019-02-11T21:39:45Z

Test build #102206 has finished for PR 23733 at commit 3c5852a.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2019-02-11T21:42:28Z

...t/resources/structured-streaming/escaped-path-2.4.0/output%20%25@%23output/_spark_metadata/0

@@ -0,0 +1,2 @@
+v1
+{"path":"file://TEMPDIR/output%20%25@%23output/part-00000-97f675a2-bb82-4201-8245-05f3dae4c372-c000.snappy.parquet","size":404,"isDir":false,"modificationTime":1549649385000,"blockReplication":1,"blockSize":33554432,"action":"add"}


TEMPDIR in this file will be replaced with the real test path in the unit test.

zsxwing · 2019-02-11T22:18:32Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala

+        if (fs.isDirectory(hdfsPath)) {
+          val metadataPath = new Path(hdfsPath, metadataDir)
+          checkEscapedMetadataPath(fs, metadataPath, sqlConf)
+          fs.exists(metadataPath)


I removed try-catch from fs.exists(metadataPath). We should not ignore random errors when failing to access metadataPath as it may make Spark ignore _spark_metadata and return wrong answers.

SparkQA · 2019-02-11T23:42:42Z

Test build #102215 has finished for PR 23733 at commit 8f408f0.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-11T23:52:04Z

Test build #102209 has finished for PR 23733 at commit 3f55c47.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-12T00:42:39Z

Test build #102213 has finished for PR 23733 at commit 7c62458.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-12T01:08:12Z

Test build #102214 has finished for PR 23733 at commit ca7bbed.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-12T01:32:58Z

Test build #102217 has finished for PR 23733 at commit 4832ce5.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-12T01:51:31Z

Test build #102218 has finished for PR 23733 at commit 6d7d1d8.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-12T02:53:28Z

Test build #102220 has finished for PR 23733 at commit 2b82f3f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gaborgsomogyi · 2019-02-12T09:16:06Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala

        } catch {
          case NonFatal(e) =>
-            logWarning(s"Error while looking for metadata directory.")
+            // We may not have access to this directory. Don't fail the query if that happens.
+            logWarning(e.getMessage, e)


This means the directory may exist but some other (maybe intermittent) exception other than FileNotFoundException came isn't it?

That's correct.

The reason I chose to ignore the error here is we may not have the permission to check the directory. For example, if I have some special chars in my user name such as foo bar. Then I try to write into my home directory /user/foo bar/a/b/c in Spark 3.0.0, it's likely I don't have access to /user/foo%20bar/a/b/c since that's in a different user home directory. Since checking the directory should be best effort and should not impact any users that don't hit this path issue, I prefer to ignore the error for safety.

This is different than try-catch in hasMetadata which is checking a sub directory in the current directory. In this case, the directory is usually accessible and the error is probably a real issue.

That's a good edge case which I've not considered. I think user with name foo%20bar or something similar which contains escaped URI parts is junk which may highlight error in some application code. Wouldn't it be better to let the user decide what to do? For example:

delete/move the dir

add right to read

ignore the escaped path check with the added config

@gaborgsomogyi Making this check best effort is safer. I don't want to force people that don't hit this issue to use the config until we remove this check in future.

@gaborgsomogyi By the way, the user name may be weird. Another more possible permission error case is an admin can set a S3 bucket to allow Spark accessing only paths matching some specified patterns, and a Spark user may not be able to change this.

gaborgsomogyi · 2019-02-12T10:01:47Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

@@ -89,8 +89,38 @@ abstract class StreamExecution(
  val resolvedCheckpointRoot = {
    val checkpointPath = new Path(checkpointRoot)
    val fs = checkpointPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
-    fs.mkdirs(checkpointPath)
-    checkpointPath.makeQualified(fs.getUri, fs.getWorkingDirectory).toUri.toString
+    if (sparkSession.conf.get(SQLConf.STREAMING_CHECKPOINT_ESCAPED_PATH_CHECK_ENABLED)


Does it worth to wrap this part in a function just like FileStreamSink.checkEscapedMetadataPath?

gaborgsomogyi · 2019-02-12T10:03:54Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

+        try {
+          fs.exists(new Path(legacyCheckpointDir))
+        } catch {
+          case NonFatal(e) =>


Same question here.

gaborgsomogyi · 2019-02-12T11:31:12Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala

+      }
+    }
+
+    withTempDir { tempDir =>


Wouldn't it be better to make withSQLConf(SQLConf.STREAMING_CHECKPOINT_ESCAPED_PATH_CHECK_ENABLED.key -> "true") explicit (I know that's the default now)?

Explicitly setting a config in a test is usually because we may change the default value in future and we don't want to break any tests.

However, for this special config, I prefer to not set it in test to make sure the default behavior is correct. If we decide to turn if off by default in future, we probably will just remove this check instead. I don't see a reason to turn this off by default and ask the user to turn it on in some cases.

Yeah, good argument and I agree.

gaborgsomogyi · 2019-02-12T11:32:37Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala

+        if (fs.isDirectory(hdfsPath)) {
+          val metadataPath = new Path(hdfsPath, metadataDir)
+          checkEscapedMetadataPath(fs, metadataPath, sqlConf)
+          fs.exists(metadataPath)


marmbrus · 2019-02-15T23:39:15Z

This is great, but I found the error a bit hard to read. What do you think about something like:

Error: we detected a possible problem with the location of your checkpoint 
and you likely need to move it before restarting this query.

Earlier version of Spark incorrectly escaped paths when writing out 
checkpoints for structured streaming.  While this was corrected in 
Spark 3.0, it appears that your query was started using an earlier 
version that incorrectly handled the checkpoint path.

Correct Checkpoint Directory: ....
Incorrect Checkpoint Directory: ...

Please move the data from the incorrect directory to the the 
correct one and then restart this query. If you believe you are 
receiving this message in error, you can disable it with (...)

SparkQA · 2019-02-19T18:30:18Z

Test build #102516 has started for PR 23733 at commit 91542ff.

brkyvz · 2019-02-20T18:36:44Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala

+      withSQLConf(SQLConf.STREAMING_CHECKPOINT_ESCAPED_PATH_CHECK_ENABLED.key -> "false") {
+        // Verify that the batch query ignores the legacy "_spark_metadata"
+        val df = spark.read.load(outputDir.getCanonicalPath)
+        assert(!(df.queryExecution.executedPlan.toString contains "MetadataLogFileIndex"))


These are amazing tests 🥇 💯

brkyvz · 2019-02-20T18:37:53Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala

+        // path issue.
+        "file://foo:b ar@bar/foo/bar",
+        "file://foo:bar@b ar/foo/bar",
+        "file://f oo:bar@bar/foo/bar").foreach { p =>


does the LocalFileSystem take an authority? Do we need to use a different file system?

@brkyvz The APIs called in containsSpecialCharsInPath don't involve FileSystems. The scheme file here can be replaced with random strings.

brkyvz

This is amazing. LGTM!

zsxwing · 2019-02-20T19:05:53Z

Thanks! Merging to master.

zsxwing · 2019-02-20T19:06:16Z

retest this please.

Triggering a new build before merging.

SparkQA · 2019-02-20T23:37:55Z

Test build #4560 has finished for PR 23733 at commit 91542ff.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

…n it contains special chars ## What changes were proposed in this pull request? When a user specifies a checkpoint location or a file sink output using a path containing special chars that need to be escaped in a path, the streaming query will store checkpoint and file sink metadata in a wrong place. In this PR, I uploaded a checkpoint that was generated by the following codes using Spark 2.4.0 to show this issue: ``` implicit val s = spark.sqlContext val input = org.apache.spark.sql.execution.streaming.MemoryStream[Int] input.addData(1, 2, 3) val q = input.toDF.writeStream.format("parquet").option("checkpointLocation", ".../chk %#chk").start(".../output %#output") q.stop() ``` Here is the structure of the directory: ``` sql/core/src/test/resources/structured-streaming/escaped-path-2.4.0 ├── chk%252520%252525%252523chk │ ├── commits │ │ └── 0 │ ├── metadata │ └── offsets │ └── 0 ├── output %#output │ └── part-00000-97f675a2-bb82-4201-8245-05f3dae4c372-c000.snappy.parquet └── output%20%25%23output └── _spark_metadata └── 0 ``` In this checkpoint, the user specified checkpoint location is `.../chk %#chk` but the real path to store the checkpoint is `.../chk%252520%252525%252523chk` (this is generated by escaping the original path three times). The user specified output path is `.../output %#output` but the path to store `_spark_metadata` is `.../output%20%25%23output/_spark_metadata` (this is generated by escaping the original path once). The data files are still in the correct path (such as `.../output %#output/part-00000-97f675a2-bb82-4201-8245-05f3dae4c372-c000.snappy.parquet`). This checkpoint will be used in unit tests in this PR. The fix is just simply removing improper `Path.toUri` calls to fix the issue. However, as the user may not read the release note and is not aware of this checkpoint location change, if they upgrade Spark without moving checkpoint to the new location, their query will just start from the scratch. In order to not surprise the users, this PR also adds a check to **detect the impacted paths and throws an error** to include the migration guide. This check can be turned off by an internal sql conf `spark.sql.streaming.checkpoint.escapedPathCheck.enabled`. Here are examples of errors that will be reported: - Streaming checkpoint error: ``` Error: we detected a possible problem with the location of your checkpoint and you likely need to move it before restarting this query. Earlier version of Spark incorrectly escaped paths when writing out checkpoints for structured streaming. While this was corrected in Spark 3.0, it appears that your query was started using an earlier version that incorrectly handled the checkpoint path. Correct Checkpoint Directory: /.../chk %#chk Incorrect Checkpoint Directory: /.../chk%252520%252525%252523chk Please move the data from the incorrect directory to the correct one, delete the incorrect directory, and then restart this query. If you believe you are receiving this message in error, you can disable it with the SQL conf spark.sql.streaming.checkpoint.escapedPathCheck.enabled. ``` - File sink error (`_spark_metadata`): ``` Error: we detected a possible problem with the location of your "_spark_metadata" directory and you likely need to move it before restarting this query. Earlier version of Spark incorrectly escaped paths when writing out the "_spark_metadata" directory for structured streaming. While this was corrected in Spark 3.0, it appears that your query was started using an earlier version that incorrectly handled the "_spark_metadata" path. Correct "_spark_metadata" Directory: /.../output %#output/_spark_metadata Incorrect "_spark_metadata" Directory: /.../output%20%25%23output/_spark_metadata Please move the data from the incorrect directory to the correct one, delete the incorrect directory, and then restart this query. If you believe you are receiving this message in error, you can disable it with the SQL conf spark.sql.streaming.checkpoint.escapedPathCheck.enabled. ``` ## How was this patch tested? The new unit tests. Closes apache#23733 from zsxwing/path-fix. Authored-by: Shixiong Zhu <zsxwing@gmail.com> Signed-off-by: Shixiong Zhu <zsxwing@gmail.com>

zsxwing added 2 commits February 4, 2019 02:10

fix

679dca2

test

3b357e5

zsxwing changed the title ~~[SPARK-26824][SS]Fix the checkpoint location and _spark_metadata when it contains special chars~~ [SPARK-26824][SS][WIP]Fix the checkpoint location and _spark_metadata when it contains special chars Feb 4, 2019

zsxwing added 4 commits February 11, 2019 10:19

migration error

3c5852a

update

3f55c47

update test

7c62458

update

ca7bbed

remove crc files

8f408f0

zsxwing commented Feb 11, 2019

View reviewed changes

update

4832ce5

zsxwing changed the title ~~[SPARK-26824][SS][WIP]Fix the checkpoint location and _spark_metadata when it contains special chars~~ [SPARK-26824][SS]Fix the checkpoint location and _spark_metadata when it contains special chars Feb 11, 2019

update

6d7d1d8

zsxwing commented Feb 11, 2019

View reviewed changes

more tests

2b82f3f

gaborgsomogyi reviewed Feb 12, 2019

View reviewed changes

gaborgsomogyi mentioned this pull request Feb 12, 2019

[SPARK-26825][SS] Fix temp checkpoint creation in cluster mode when default filesystem is not local. #23764

Closed

update error message

91542ff

brkyvz reviewed Feb 20, 2019

View reviewed changes

brkyvz approved these changes Feb 20, 2019

View reviewed changes

asfgit closed this in 77b99af Feb 20, 2019

zsxwing deleted the path-fix branch February 20, 2019 23:48

gaborgsomogyi mentioned this pull request Nov 15, 2019

[SPARK-29779][CORE] Compact old event log files and cleanup #26416

Closed

		@@ -0,0 +1,2 @@
		v1
		{"path":"file://TEMPDIR/output%20%25@%23output/part-00000-97f675a2-bb82-4201-8245-05f3dae4c372-c000.snappy.parquet","size":404,"isDir":false,"modificationTime":1549649385000,"blockReplication":1,"blockSize":33554432,"action":"add"}

[SPARK-26824][SS]Fix the checkpoint location and _spark_metadata when it contains special chars #23733

[SPARK-26824][SS]Fix the checkpoint location and _spark_metadata when it contains special chars #23733

Uh oh!

Conversation

zsxwing commented Feb 4, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

zsxwing commented Feb 4, 2019

Uh oh!

SparkQA commented Feb 4, 2019

Uh oh!

zsxwing commented Feb 4, 2019

Uh oh!

SparkQA commented Feb 11, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Feb 11, 2019

Uh oh!

SparkQA commented Feb 11, 2019

Uh oh!

SparkQA commented Feb 12, 2019

Uh oh!

SparkQA commented Feb 12, 2019

Uh oh!

SparkQA commented Feb 12, 2019

Uh oh!

SparkQA commented Feb 12, 2019

Uh oh!

SparkQA commented Feb 12, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zsxwing Feb 12, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

marmbrus commented Feb 15, 2019

Uh oh!

SparkQA commented Feb 19, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

brkyvz left a comment

Choose a reason for hiding this comment

Uh oh!

zsxwing commented Feb 20, 2019

Uh oh!

zsxwing commented Feb 20, 2019

Uh oh!

SparkQA commented Feb 20, 2019

Uh oh!

zsxwing commented Feb 4, 2019 •

edited

Loading

zsxwing Feb 12, 2019 •

edited

Loading