[SPARK-38323][SQL][Streaming] Support the hidden file metadata in Streaming #35676

Yaohua628 · 2022-02-28T07:50:10Z

What changes were proposed in this pull request?

Support the hidden file source metadata in streaming, you could do the same thing as batch read/write as follows:

spark
  .readStream
  ...
  .select("*", "_metadata")
  .writeStream
  ...
  .start()

Why are the changes needed?

Add more support to the hidden file metadata feature. Before this PR, querying the hidden file metadata struct _metadata will fail using readStream, writeStream streaming APIs.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Add a new UT

Yaohua628 · 2022-02-28T07:53:19Z

@cloud-fan appreciate it if you can take a look, thanks!

cloud-fan · 2022-02-28T11:50:04Z

Can you fill the What changes were proposed in this pull request? section and fix merge conflicts? The code change looks good. Also cc @HeartSaVioR

AmplabJenkins · 2022-03-01T03:50:34Z

Can one of the admins verify this patch?

cloud-fan · 2022-03-01T13:09:26Z

thanks, merging to master!

HeartSaVioR · 2022-03-03T02:48:35Z

...core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala

+      )
+
+      // SELECT * will have: name, age, info, _metadata of /source/new-streaming-data
+      assert(newDF.select("*").columns.toSet == Set("name", "age", "info", "_metadata"))


@Yaohua628

Sorry for the post review. I haven't had a time to review this in time.

Just to make clear, select("*").show() should not expose a hidden column, right? Since you've included "_metadata" from the list of columns so I would like to double confirm that it is not user facing.

And given we include the new column, dropDuplicate without explicitly mentioning columns in streaming query would be broken. state schema would somehow include the hidden column in the schema, whereas state schema from older version of checkpoint does not include the hidden column (as they didn't exist).

We should test it, and if it fall into the case, we should mention this in the migration guide, or make this configurable and by default turn off. (We did this for adding Kafka header - #22282)

cc. @cloud-fan

Yep, select("*") won't expose the hidden file metadata column.

But here what I did is readStream and explicitly selecting * and _metadata (here) and writeStream to a target table /target/new-streaming-data, so the target table itself will have a column called _metadata.

Ah OK my bad. That is just checking the output.

dropDuplicate() still remains a question. How we deal with this? We removed the column by default for Kafka header to not break compatibility with dropDuplicate(), but not sure we would like to add more config.

I see! I can test with dropDuplicates() and see how it goes

Yaohua628 · 2022-03-03T09:21:23Z

After manual testing locally, changes of _metadata in streaming won't cause any backward-compatible issues for dropDuplicates():

with Spark 3.2: readStream...select("*").dropDuplicates().writeStream...option("checkpointLocation", ".../checkpoint")
verify results (duplicated data did get dropped), verify checkpoint dir (commits, offsets, sources, state)
add some new files including duplicates
with Spark master (3.3 with _metadata feature) with the same checkpoint location: readStream...select("*").dropDuplicates().writeStream...option("checkpointLocation", ".../checkpoint")
verify results (duplicated data did get dropped), verify checkpoint dir (commits, offsets, sources, state)

HeartSaVioR · 2022-03-03T09:43:49Z

Thanks for checking it manually! We are good to go.

### What changes were proposed in this pull request? We added the support to query the `_metadata` column with a file-based streaming source: #35676. We propose to use `transformUp` instead of `match` when pattern matching the `dataPlan` in `MicroBatchExecution` `runBatch` method in this PR. It is fine for `FileStreamSource` because `FileStreamSource` always returns one `LogicalRelation` node (https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala#L247). But the proposed change will make the logic robust and we really should not rely on the upstream source to return a desired plan. In addition, the proposed change could also make `_metadata` work if someone wants to customize `FileStreamSource` `getBatch`. ### Why are the changes needed? Robust ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests Closes #36801 from Yaohua628/spark-39404. Authored-by: yaohua <yaohua.zhao@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

### What changes were proposed in this pull request? We added the support to query the `_metadata` column with a file-based streaming source: apache#35676. We propose to use `transformUp` instead of `match` when pattern matching the `dataPlan` in `MicroBatchExecution` `runBatch` method in this PR. It is fine for `FileStreamSource` because `FileStreamSource` always returns one `LogicalRelation` node (https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala#L247). But the proposed change will make the logic robust and we really should not rely on the upstream source to return a desired plan. In addition, the proposed change could also make `_metadata` work if someone wants to customize `FileStreamSource` `getBatch`. ### Why are the changes needed? Robust ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests Closes apache#36801 from Yaohua628/spark-39404. Authored-by: yaohua <yaohua.zhao@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

### What changes were proposed in this pull request? (This cherry-picks #36801) We added the support to query the `_metadata` column with a file-based streaming source: #35676. We propose to use `transformUp` instead of `match` when pattern matching the `dataPlan` in `MicroBatchExecution` `runBatch` method in this PR. It is fine for `FileStreamSource` because `FileStreamSource` always returns one `LogicalRelation` node (https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala#L247). But the proposed change will make the logic robust and we really should not rely on the upstream source to return a desired plan. In addition, the proposed change could also make `_metadata` work if someone wants to customize `FileStreamSource` `getBatch`. ### Why are the changes needed? Robust ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests Closes #38337 from Yaohua628/spark-39404-3-3. Authored-by: yaohua <yaohua.zhao@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

streaming for file metadata

a215439

github-actions bot added SQL STRUCTURED STREAMING labels Feb 28, 2022

Yaohua628 added 2 commits March 1, 2022 11:36

Merge branch 'master' into spark-38323

66de847

nit

ae9959c

cloud-fan approved these changes Mar 1, 2022

View reviewed changes

cloud-fan closed this in ad4e5a6 Mar 1, 2022

HeartSaVioR reviewed Mar 3, 2022

View reviewed changes

Yaohua628 mentioned this pull request Jun 8, 2022

[SPARK-39404][SS] Minor fix for querying _metadata in streaming #36801

Closed

HeartSaVioR mentioned this pull request Oct 22, 2022

[SPARK-39404][SS][3.3] Minor fix for querying _metadata in streaming #38337

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[SPARK-38323][SQL][Streaming] Support the hidden file metadata in Streaming #35676

[SPARK-38323][SQL][Streaming] Support the hidden file metadata in Streaming #35676

Uh oh!

Yaohua628 commented Feb 28, 2022 •

edited

Loading

Uh oh!

Yaohua628 commented Feb 28, 2022

Uh oh!

cloud-fan commented Feb 28, 2022

Uh oh!

AmplabJenkins commented Mar 1, 2022

Uh oh!

cloud-fan commented Mar 1, 2022

Uh oh!

HeartSaVioR Mar 3, 2022 •

edited

Loading

Uh oh!

Yaohua628 Mar 3, 2022

Uh oh!

HeartSaVioR Mar 3, 2022

Uh oh!

Yaohua628 Mar 3, 2022

Uh oh!

Yaohua628 commented Mar 3, 2022

Uh oh!

HeartSaVioR commented Mar 3, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

[SPARK-38323][SQL][Streaming] Support the hidden file metadata in Streaming #35676

[SPARK-38323][SQL][Streaming] Support the hidden file metadata in Streaming #35676

Uh oh!

Conversation

Yaohua628 commented Feb 28, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

Yaohua628 commented Feb 28, 2022

Uh oh!

cloud-fan commented Feb 28, 2022

Uh oh!

AmplabJenkins commented Mar 1, 2022

Uh oh!

cloud-fan commented Mar 1, 2022

Uh oh!

HeartSaVioR Mar 3, 2022 • 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!

Yaohua628 Mar 3, 2022

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Mar 3, 2022

Choose a reason for hiding this comment

Uh oh!

Yaohua628 Mar 3, 2022

Choose a reason for hiding this comment

Uh oh!

Yaohua628 commented Mar 3, 2022

Uh oh!

HeartSaVioR commented Mar 3, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Yaohua628 commented Feb 28, 2022 •

edited

Loading

HeartSaVioR Mar 3, 2022 •

edited

Loading