This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new ceeca31aaf6e [SPARK-58605][SQL] Verify parent-archive _metadata for
archive reads
ceeca31aaf6e is described below
commit ceeca31aaf6ecb7a2d582a223b53291e804fd4e8
Author: akshatshenoi-db <[email protected]>
AuthorDate: Thu Aug 6 19:18:39 2026 +0800
[SPARK-58605][SQL] Verify parent-archive _metadata for archive reads
### What changes were proposed in this pull request?
Adds tests asserting that reading an archive exposes the **parent archive
file's** values in the `_metadata` column (`file_path`, `file_name`,
`file_size`, `file_modification_time`, `file_block_start`, `file_block_length`).
An archive is read as a single non-splittable `PartitionedFile`, so the
generic `FileFormat` metadata extractors already produce the parent archive's
values (`file_block_start` = 0, `file_block_length` = `file_size` = the
archive's size on disk). This PR is therefore **test-only**: no production
change was needed. The tests pin the behavior so it cannot silently regress.
Inner-file `_metadata` is explicitly a **non-goal**: because the archive is
a single split, the `_metadata` mechanism operates at `PartitionedFile`
granularity and cannot attribute a row to its originating inner entry.
This follows the archive-reader series: SPARK-57135 / SPARK-57321 (CSV),
SPARK-57419 (JSON), SPARK-57478 (text), SPARK-57479 (XML), SPARK-57481 (Avro),
SPARK-58382 (binaryFile), SPARK-57705 (zip), SPARK-58246 (7z).
### Why are the changes needed?
The parent-archive `_metadata` contract was entirely untested. A future
change to file splitting or to the metadata extractors could regress it
unnoticed, and users reading archives rely on `_metadata` to identify the
source archive file.
### Does this PR introduce any user-facing change?
No. Test-only; archive reading remains gated by
`spark.sql.files.archive.reader.enabled` (default false).
### How was this patch tested?
A new shared test in `ArchiveReadSuiteBase`, which runs for every format x
container suite (csv/json/xml/avro x tar/zip/7z), asserting every row of a
multi-entry archive carries the same parent-archive values, with
`file_block_start = 0` and `file_block_length = file_size = ` the archive's
size.
Plus the parallel test in `TextArchiveReadBase` and
`BinaryFileArchiveReadBase`, which do not extend `ArchiveReadSuiteBase` (their
row shapes differ). For binaryFile with `wholeFile=false` the test additionally
pins that `_metadata` stays parent-only even though the `path`/`length`
**data** columns are sourced per entry.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #57805 from akshatshenoi-db/archive-metadata.
Authored-by: akshatshenoi-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit cc7fac39d303e8f819edb20df165b66c53467177)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../datasources/ArchiveReadSuiteBase.scala | 37 ++++++++++++++++++++++
.../datasources/BinaryFileArchiveReadBase.scala | 31 ++++++++++++++++++
.../datasources/TextArchiveReadBase.scala | 29 +++++++++++++++++
3 files changed, 97 insertions(+)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
index ab448a4f2c80..d10476c52dea 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
@@ -637,6 +637,43 @@ trait ArchiveReadSuiteBase extends QueryTest with
SharedSparkSession {
}
}
+ // ----- shared parent-archive _metadata test
--------------------------------
+
+ test("_metadata exposes the parent archive file's values, identical for
every row") {
+ archiveExtensions.foreach { ext =>
+ withArchiveFile(ext) { archive =>
+ // Multiple entries, each multiple rows: the archive is one
non-splittable PartitionedFile,
+ // so every row must carry the same parent-archive metadata (not any
inner entry's).
+ val parts = Seq(sampleDf((1, "Alice"), (2, "Bob")), sampleDf((3,
"Carol"), (4, "Dan")))
+ writeArchive(
+ archive, parts.zipWithIndex.map { case (p, i) => entryName(i) ->
encodeFile(p) })
+
+ val rows = read(archive.getCanonicalPath)
+ .select("_metadata.file_path", "_metadata.file_name",
"_metadata.file_size",
+ "_metadata.file_block_start", "_metadata.file_block_length",
+ "_metadata.file_modification_time")
+ .collect()
+ assert(rows.length == parts.map(_.count()).sum,
+ s"expected one row per input record, got ${rows.length}")
+
+ val fileSize = archive.length()
+ rows.foreach { r =>
+ assert(r.getString(0).endsWith(archive.getName) &&
!r.getString(0).contains(entryName(0)),
+ s"file_path should be the archive file, got ${r.getString(0)}")
+ assert(r.getString(1) == archive.getName, s"file_name mismatch:
${r.getString(1)}")
+ assert(r.getLong(2) == fileSize, s"file_size mismatch:
${r.getLong(2)} != $fileSize")
+ assert(r.getLong(3) == 0L, s"file_block_start should be 0, got
${r.getLong(3)}")
+ assert(r.getLong(4) == fileSize,
+ s"file_block_length should be the archive size, got
${r.getLong(4)}")
+ assert(r.getAs[java.sql.Timestamp](5).getTime ==
archive.lastModified(),
+ "file_modification_time should be the archive's mtime")
+ }
+ assert(rows.map(_.toSeq).distinct.length == 1,
+ "every row must carry the same parent-archive metadata")
+ }
+ }
+ }
+
// ----- shared complex-type test (run when `supportsComplexTypes`)
----------
if (supportsComplexTypes) {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
index c115f9bafa53..e23c61a8236a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
@@ -107,6 +107,37 @@ trait BinaryFileArchiveReadBase extends QueryTest with
SharedSparkSession {
}
}
+ test("wholeFile=false _metadata exposes the parent archive's values for
every row") {
+ archiveExtensions.foreach { ext =>
+ withArchiveFile(ext) { archive =>
+ writeArchive(archive, Seq("a.bin" -> bytes("aaa"), "b.bin" ->
bytes("bbbb")))
+ val rows = read(archive.getCanonicalPath, Map("wholeFile" -> "false"))
+ .select("_metadata.file_path", "_metadata.file_name",
"_metadata.file_size",
+ "_metadata.file_block_start", "_metadata.file_block_length",
+ "_metadata.file_modification_time")
+ .collect()
+ assert(rows.length == 2)
+
+ val fileSize = archive.length()
+ rows.foreach { r =>
+ // The `path`/`length` data columns are per entry here, but
_metadata stays parent-only:
+ // it is derived from the single PartitionedFile.
+ assert(r.getString(0).endsWith(archive.getName) &&
!r.getString(0).contains("!/"),
+ s"file_path should be the archive file, got ${r.getString(0)}")
+ assert(r.getString(1) == archive.getName, s"file_name mismatch:
${r.getString(1)}")
+ assert(r.getLong(2) == fileSize, s"file_size mismatch:
${r.getLong(2)} != $fileSize")
+ assert(r.getLong(3) == 0L, s"file_block_start should be 0, got
${r.getLong(3)}")
+ assert(r.getLong(4) == fileSize,
+ s"file_block_length should be the archive size, got
${r.getLong(4)}")
+ assert(r.getAs[java.sql.Timestamp](5).getTime ==
archive.lastModified(),
+ "file_modification_time should be the archive's mtime")
+ }
+ assert(rows.map(_.toSeq).distinct.length == 1,
+ "every row must carry the same parent-archive metadata")
+ }
+ }
+ }
+
test("wholeFile=false on an empty archive yields no rows") {
withArchiveFile() { archive =>
writeArchive(archive, Seq.empty)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TextArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TextArchiveReadBase.scala
index 1204c5a71b04..03c8b95d23c8 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TextArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TextArchiveReadBase.scala
@@ -155,6 +155,35 @@ trait TextArchiveReadBase extends QueryTest with
SharedSparkSession {
}
}
+ test("_metadata exposes the parent archive file's values, identical for
every row") {
+ archiveExtensions.foreach { ext =>
+ withArchiveFile(ext) { archive =>
+ writeArchive(archive, Seq(
+ "a.txt" -> textBytes("l1\nl2\n"), "b.txt" -> textBytes("l3\nl4\n")))
+ val rows = read(archive.getCanonicalPath)
+ .select("_metadata.file_path", "_metadata.file_name",
"_metadata.file_size",
+ "_metadata.file_block_start", "_metadata.file_block_length",
+ "_metadata.file_modification_time")
+ .collect()
+ assert(rows.length == 4)
+ val fileSize = archive.length()
+ rows.foreach { r =>
+ assert(r.getString(0).endsWith(archive.getName) &&
!r.getString(0).contains("a.txt"),
+ s"file_path should be the archive file, got ${r.getString(0)}")
+ assert(r.getString(1) == archive.getName, s"file_name mismatch:
${r.getString(1)}")
+ assert(r.getLong(2) == fileSize, s"file_size mismatch:
${r.getLong(2)} != $fileSize")
+ assert(r.getLong(3) == 0L, s"file_block_start should be 0, got
${r.getLong(3)}")
+ assert(r.getLong(4) == fileSize,
+ s"file_block_length should be the archive size, got
${r.getLong(4)}")
+ assert(r.getAs[java.sql.Timestamp](5).getTime ==
archive.lastModified(),
+ "file_modification_time should be the archive's mtime")
+ }
+ assert(rows.map(_.toSeq).distinct.length == 1,
+ "every row must carry the same parent-archive metadata")
+ }
+ }
+ }
+
Seq(true, false).foreach { ignoreCorrupt =>
test(s"ignoreCorruptFiles=$ignoreCorrupt controls whether a corrupt
archive is skipped") {
withArchiveFile(corruptArchiveExtension) { archive =>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
