yihua commented on code in PR #20079:
URL: https://github.com/apache/hudi/pull/20079#discussion_r4188124086
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java:
##########
@@ -196,6 +198,25 @@ public void close() {
}
}
+ /**
+ * Sets the Hadoop read options of a reader built with {@code
ParquetReader.Builder(InputFile)} from the
+ * file's {@link Configuration}, as {@code ParquetReader.Builder(Path)}
followed by {@code withConf} does,
+ * without creating a new {@link Configuration}. {@code withConf} rebuilds
the read options from the
+ * configuration; the decryption properties are the only option that takes
the file path, which
+ * {@code withConf} drops, so they are resolved here with it. On parquet
1.14+ the {@code InputFile}
+ * constructor alone builds plain {@code ParquetReadOptions}, which never
consult the
+ * {@code parquet.crypto.factory.class} decryption factory.
+ */
+ public static <T> ParquetReader.Builder<T>
withHadoopReadOptions(ParquetReader.Builder<T> builder, HadoopInputFile file) {
Review Comment:
Done, the javadoc now says `withConf` rebuilds the options from the
builder's null `path` field, so parquet would hand the decryption factory a
null path.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala:
##########
@@ -80,9 +81,21 @@ class ParquetSchemaEvolutionUtils(sharedConf: Configuration,
protected var typeChangeInfos: java.util.Map[Integer, Pair[DataType,
DataType]] = null
- def getHadoopConfClone(footerFileMetaData: FileMetaData,
enableVectorizedReader: Boolean): Configuration = {
- // Clone new conf
- val hadoopAttemptConf = new Configuration(sharedConf)
+ /**
+ * Returns the configuration to read the file with: the read configuration
itself when the file needs no
+ * keys of its own, otherwise a copy with the file's requested schema. Pass
`writable` when the caller sets
+ * keys on the returned configuration. The read configuration is never
modified, since other readers may
+ * share it.
+ */
+ def getFileReadConf(footerFileMetaData: FileMetaData,
enableVectorizedReader: Boolean, writable: Boolean): Configuration = {
+ // A JobConf, so the task attempt context built on it does not copy it
again
+ var fileReadConf: Configuration = if (writable) new JobConf(readConf) else
readConf
Review Comment:
Addressed: `getFileReadConf` now takes the pushed filter and sets it on the
per-file copy itself, so readers never write to the returned configuration and
the `writable` flag is gone.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkParquetReaderBase.scala:
##########
@@ -64,7 +65,8 @@ abstract class SparkParquetReaderBase(enableVectorizedReader:
Boolean,
filters: Seq[Filter],
storageConf: StorageConfiguration[Configuration],
tableSchemaOpt:
util.Option[org.apache.parquet.schema.MessageType] = util.Option.empty()):
Iterator[InternalRow] = {
- val conf = storageConf.unwrapCopy()
+ // A JobConf, so the task attempt context doRead builds on it reuses it
instead of copying it again
+ val conf = new JobConf(storageConf.unwrap())
Review Comment:
Done, both `JobConf` comments now name `JobContextImpl` reusing a `JobConf`
as what the saving depends on.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
