You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/13 10:44:09 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6923-8da02184af3d0734d30732a3f45b8b81b1d55c96 created (now 987ee7397f)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6923-8da02184af3d0734d30732a3f45b8b81b1d55c96
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 987ee7397f perf: read native-written Parquet footers without parsing Hadoop's configuration (#6923)
This branch includes the following new commits:
new 987ee7397f perf: read native-written Parquet footers without parsing Hadoop's configuration (#6923)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: perf: read native-written Parquet footers without parsing Hadoop's configuration (#6923)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6923-8da02184af3d0734d30732a3f45b8b81b1d55c96
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 987ee7397f081f8f360803c9f593308eedde0a56
Author: Andy Grove <ag...@apache.org>
AuthorDate: Tue Oct 13 10:43:46 2026 +0000
perf: read native-written Parquet footers without parsing Hadoop's configuration (#6923)
After a native Iceberg write, every data file's footer is re-read so that
its metrics come from iceberg-java. ParquetFileReader.open(InputFile)
builds its read options from a new Hadoop Configuration on parquet-mr
1.14 and later, which parses Hadoop's default XML resources once for
every data file. Read the footer with parquet-mr's default options built
from a PlainParquetConfiguration instead, and keep open(InputFile) on
older parquet-mr, whose defaults are built without Hadoop.
---
.../apache/comet/iceberg/IcebergReflection.scala | 65 +++++++++++++++++++---
.../comet/iceberg/IcebergReflectionSuite.scala | 19 +++++++
2 files changed, 76 insertions(+), 8 deletions(-)
diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index 9662663139..354ff37e21 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -19,7 +19,7 @@
package org.apache.comet.iceberg
-import java.lang.reflect.Method
+import java.lang.reflect.{Constructor, Method}
import java.util.concurrent.ConcurrentHashMap
import scala.util.control.NonFatal
@@ -2152,18 +2152,41 @@ object IcebergReflection extends Logging {
}
/**
- * The three parquet-mr-bridging members the footer re-read depends on. Resolved together, per
- * call, rather than memoized in JVM-lifetime lazy vals: `executorReflectionUnresolved` keys its
+ * The parquet-mr-bridging members the footer re-read depends on. Resolved together, per call,
+ * rather than memoized in JVM-lifetime lazy vals: `executorReflectionUnresolved` keys its
* verdict on the current Iceberg class loader, and a probe re-run after a classpath change must
* not be satisfied by `Method`s resolved against a previous Iceberg. Resolution is a per-task
* cost (once per `rebuildDataFilesWithJavaMetrics` call), not per-file.
*/
- private case class ParquetFooterReflection(
+ private[iceberg] case class ParquetFooterReflection(
footerMetrics: Method,
parquetIOFile: Method,
- parquetFileReaderOpen: Method)
+ parquetFileReaderOpen: Method,
+ readOptions: Option[FooterReadOptions])
- private def resolveParquetFooterReflection(): ParquetFooterReflection = {
+ /**
+ * Opens a file with parquet-mr's default read options, built from a `PlainParquetConfiguration`
+ * rather than from a Hadoop `Configuration`.
+ *
+ * On parquet-mr that has `ParquetConfiguration` (1.14 and later),
+ * `ParquetFileReader.open(InputFile)` builds its options from a new Hadoop `Configuration` and
+ * reads them back, which parses Hadoop's default XML resources on every call, so once for every
+ * data file a task writes. The options govern filtering, decryption, allocation, and how string
+ * statistics from old parquet-mr writers are read, none of which applies to a footer the native
+ * writer has just written. Earlier parquet-mr builds its defaults without Hadoop, so there the
+ * footer is still read through `open(InputFile)`.
+ */
+ private[iceberg] case class FooterReadOptions(
+ open: Method,
+ builder: Method,
+ build: Method,
+ plainConfiguration: Constructor[_]) {
+
+ def newOptions(): AnyRef =
+ build.invoke(builder.invoke(null, plainConfiguration.newInstance().asInstanceOf[AnyRef]))
+ }
+
+ private[iceberg] def resolveParquetFooterReflection(): ParquetFooterReflection = {
// `ParquetUtil.footerMetrics(ParquetMetadata, Stream[FieldMetrics], MetricsConfig,
// NameMapping)`. In `iceberg-spark-runtime` jars parquet-mr is shaded, so the
// `ParquetMetadata` parameter type is the relocated class -- resolved from the method itself
@@ -2183,12 +2206,38 @@ object IcebergReflection extends Logging {
parquetMetadataClass.getName
.replace("hadoop.metadata.ParquetMetadata", "hadoop.ParquetFileReader"))
val parquetFileReaderOpen = readerClass.getMethod("open", parquetIOFile.getReturnType)
- ParquetFooterReflection(footerMetrics, parquetIOFile, parquetFileReaderOpen)
+ ParquetFooterReflection(
+ footerMetrics,
+ parquetIOFile,
+ parquetFileReaderOpen,
+ resolveFooterReadOptions(readerClass, parquetIOFile.getReturnType))
+ }
+
+ /**
+ * The members [[FooterReadOptions]] needs, or `None` on parquet-mr without
+ * `ParquetConfiguration`. They are looked up next to `ParquetFileReader`, which lives in the
+ * `hadoop` package of a possibly relocated parquet-mr.
+ */
+ private def resolveFooterReadOptions(
+ readerClass: Class[_],
+ inputFileClass: Class[_]): Option[FooterReadOptions] = {
+ val parquetPackage = readerClass.getName.stripSuffix("hadoop.ParquetFileReader")
+ for {
+ configurationClass <- tryLoadClass(s"${parquetPackage}conf.ParquetConfiguration")
+ plainConfigurationClass <- tryLoadClass(s"${parquetPackage}conf.PlainParquetConfiguration")
+ optionsClass <- tryLoadClass(s"${parquetPackage}ParquetReadOptions")
+ open <- findMethod(readerClass, "open", inputFileClass, optionsClass)
+ builder <- findMethod(optionsClass, "builder", configurationClass)
+ build <- findMethod(builder.getReturnType, "build")
+ } yield FooterReadOptions(open, builder, build, plainConfigurationClass.getConstructor())
}
private def readParquetFooter(parquet: ParquetFooterReflection, inputFile: AnyRef): AnyRef = {
val parquetInputFile = parquet.parquetIOFile.invoke(null, inputFile)
- val reader = parquet.parquetFileReaderOpen.invoke(null, parquetInputFile)
+ val reader = parquet.readOptions match {
+ case Some(options) => options.open.invoke(null, parquetInputFile, options.newOptions())
+ case None => parquet.parquetFileReaderOpen.invoke(null, parquetInputFile)
+ }
try {
getMethod(reader.getClass, "getFooter").invoke(reader)
} finally {
diff --git a/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala b/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala
index f5f3a75f49..7f1cfa35e8 100644
--- a/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala
@@ -280,6 +280,25 @@ class IcebergReflectionSuite extends AnyFunSuite {
IcebergReflection.executorReflectionUnresolved)
}
+ test("footer re-reads build parquet-mr's read options without a Hadoop Configuration") {
+ // `ParquetFileReader.open(InputFile)` builds its read options from a new Hadoop
+ // `Configuration` on parquet-mr 1.14 and later, and parsing that configuration's XML
+ // resources was most of the cost of rebuilding a native-written file's metrics.
+ IcebergReflection.resolveParquetFooterReflection().readOptions match {
+ case Some(readOptions) =>
+ val options = readOptions.newOptions()
+ val configuration = options.getClass.getMethod("getConfiguration").invoke(options)
+ assert(configuration.getClass.getSimpleName == "PlainParquetConfiguration")
+ case None =>
+ // Only parquet-mr without `ParquetConfiguration` keeps `open(InputFile)`, whose
+ // defaults are built without Hadoop.
+ intercept[ClassNotFoundException] {
+ IcebergReflection.loadClass(
+ "org.apache.iceberg.shaded.org.apache.parquet.conf.ParquetConfiguration")
+ }
+ }
+ }
+
/** Schema the transform tests below partition on, one column per transform source type. */
private val transformSchema = new Schema(
Types.NestedField.optional(1, "id", Types.LongType.get()),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org