This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 1c30d058c579 perf(spark): stop per-file Hadoop configuration parsing
and copies on the parquet read path (#20079)
1c30d058c579 is described below
commit 1c30d058c57978dd77abf423c6209579797d7c66
Author: Y Ethan Guo <[email protected]>
AuthorDate: Mon Oct 5 20:54:21 2026 -0700
perf(spark): stop per-file Hadoop configuration parsing and copies on the
parquet read path (#20079)
---
.../hudi/io/storage/HoodieSparkParquetReader.java | 26 ++-
.../org/apache/hudi/common/util/ParquetUtils.java | 22 +++
.../io/storage/hadoop/HoodieAvroParquetReader.java | 9 +-
.../apache/parquet/avro/HoodieAvroReadSupport.java | 2 +-
.../testutils/FooterKeyDecryptionFactory.java | 96 +++++++++++
.../HadoopDefaultResourceLoadCounter.java | 68 ++++++++
.../hadoop/TestHoodieAvroParquetReader.java | 100 ++++++++++++
.../parquet/ParquetSchemaEvolutionUtils.scala | 46 ++++--
.../parquet/SparkParquetReaderBase.scala | 5 +-
.../parquet/TestParquetSchemaEvolutionUtils.scala | 51 +++++-
.../io/storage/TestHoodieSparkParquetReader.java | 133 +++++++++++++++
.../io/storage/TestParquetReadOptionsParity.java | 148 +++++++++++++++++
.../hudi/functional/TestBasicSchemaEvolution.scala | 2 +-
.../io/storage/ParquetPerFileReadBenchmark.scala | 179 +++++++++++++++++++++
.../Spark3LegacyHoodieParquetFileFormat.scala | 2 +-
.../datasources/parquet/Spark33ParquetReader.scala | 8 +-
.../datasources/parquet/Spark34ParquetReader.scala | 8 +-
.../datasources/parquet/Spark35ParquetReader.scala | 8 +-
.../Spark40LegacyHoodieParquetFileFormat.scala | 2 +-
.../datasources/parquet/Spark40ParquetReader.scala | 8 +-
.../Spark41LegacyHoodieParquetFileFormat.scala | 2 +-
.../datasources/parquet/Spark41ParquetReader.scala | 8 +-
.../Spark42LegacyHoodieParquetFileFormat.scala | 2 +-
.../datasources/parquet/Spark42ParquetReader.scala | 8 +-
24 files changed, 876 insertions(+), 67 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkParquetReader.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkParquetReader.java
index 40497ea01fa8..d84c62f168b6 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkParquetReader.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkParquetReader.java
@@ -44,6 +44,8 @@ import org.apache.parquet.filter2.predicate.FilterApi;
import org.apache.parquet.filter2.predicate.FilterPredicate;
import org.apache.parquet.hadoop.ParquetInputFormat;
import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.api.ReadSupport;
+import org.apache.parquet.hadoop.util.HadoopInputFile;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.SchemaRepair;
import org.apache.spark.sql.HoodieInternalRowUtils;
@@ -209,8 +211,8 @@ public class HoodieSparkParquetReader implements
HoodieSparkFileReader {
Option$.MODULE$.empty(), true, true,
rebaseDateSpec,
messageSchema);
- ParquetReader<InternalRow> reader = ParquetReader.builder(readSupport, new
Path(path.toUri()))
- .withConf(storage.getConf().unwrapAs(Configuration.class))
+ HadoopInputFile inputFile = HadoopInputFile.fromPath(new
Path(path.toUri()), storage.getConf().unwrapAs(Configuration.class));
+ ParquetReader<InternalRow> reader = ParquetUtils.withHadoopReadOptions(new
InputFileReaderBuilder(readSupport, inputFile), inputFile)
.build();
UnsafeProjection projection = evolution.generateUnsafeProjection();
ParquetReaderIterator<InternalRow> parquetReaderIterator = new
ParquetReaderIterator<>(reader);
@@ -275,4 +277,24 @@ public class HoodieSparkParquetReader implements
HoodieSparkFileReader {
public long getTotalRecords() {
return parquetUtils.getRowCount(storage, path);
}
+
+ /**
+ * Builds the reader from a {@link HadoopInputFile}, whose {@link
Configuration} becomes the reader's
+ * configuration (see {@link ParquetUtils#withHadoopReadOptions}). {@code
ParquetReader.builder(ReadSupport, Path)}
+ * creates a new {@link Configuration} and loads the Hadoop default
resources for every file, even when
+ * {@code withConf} replaces it.
+ */
+ private static class InputFileReaderBuilder extends
ParquetReader.Builder<InternalRow> {
+ private final ReadSupport<InternalRow> readSupport;
+
+ InputFileReaderBuilder(ReadSupport<InternalRow> readSupport,
HadoopInputFile file) {
+ super(file);
+ this.readSupport = readSupport;
+ }
+
+ @Override
+ protected ReadSupport<InternalRow> getReadSupport() {
+ return readSupport;
+ }
+ }
}
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java
index 87961f8223cf..90f0dc64952f 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java
@@ -54,6 +54,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.parquet.avro.AvroParquetReader;
import org.apache.parquet.avro.AvroReadSupport;
import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.crypto.DecryptionPropertiesFactory;
import org.apache.parquet.format.converter.ParquetMetadataConverter;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.ParquetReader;
@@ -61,6 +62,7 @@ import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.hadoop.util.HadoopInputFile;
import org.apache.parquet.io.api.Binary;
import org.apache.parquet.schema.DecimalMetadata;
import org.apache.parquet.schema.MessageType;
@@ -197,6 +199,26 @@ public class ParquetUtils extends FileFormatUtils {
}
}
+ /**
+ * 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 and the builder's {@code path} field, which the {@code
InputFile} constructor leaves null.
+ * The decryption properties are the only option that takes the file path,
so parquet would hand the
+ * decryption factory a null path; they are resolved here with the file's
path instead. 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) {
+ Configuration conf = file.getConfiguration();
+ builder.withConf(conf);
+ DecryptionPropertiesFactory decryptionFactory =
DecryptionPropertiesFactory.loadFactory(conf);
+ if (decryptionFactory != null) {
+
builder.withDecryption(decryptionFactory.getFileDecryptionProperties(conf,
file.getPath()));
+ }
+ return builder;
+ }
+
public static ParquetMetadata readMetadata(HoodieStorage storage,
StoragePath parquetFilePath) {
return readMetadata(storage, parquetFilePath, NO_FILTER);
}
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroParquetReader.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroParquetReader.java
index 06df845ddfe4..118e7c63856b 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroParquetReader.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroParquetReader.java
@@ -31,6 +31,7 @@ import org.apache.hudi.common.util.FileFormatUtils;
import org.apache.hudi.common.util.HoodieAvroParquetReaderIterator;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ParquetReaderIterator;
+import org.apache.hudi.common.util.ParquetUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.collection.CloseableMappingIterator;
import org.apache.hudi.common.util.collection.Pair;
@@ -44,11 +45,13 @@ import org.apache.hudi.storage.StoragePath;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
import org.apache.parquet.avro.AvroReadSupport;
import org.apache.parquet.avro.AvroSchemaConverter;
import org.apache.parquet.avro.HoodieAvroParquetReaderBuilder;
import org.apache.parquet.hadoop.ParquetInputFormat;
import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.util.HadoopInputFile;
import java.io.IOException;
import java.util.ArrayList;
@@ -199,10 +202,10 @@ public class HoodieAvroParquetReader extends
HoodieAvroFileReader {
AvroReadSupport.setAvroReadSchema(hadoopConf, readSchema.toAvroSchema());
AvroReadSupport.setRequestedProjection(hadoopConf,
readSchema.toAvroSchema());
}
+ HadoopInputFile inputFile = HadoopInputFile.fromPath(new
Path(path.toUri()), hadoopConf);
ParquetReader<IndexedRecord> reader =
- new HoodieAvroParquetReaderBuilder<IndexedRecord>(path)
-
.withTableSchema(getAvroSchemaConverter(hadoopConf).convert(readSchema))
- .withConf(hadoopConf)
+ ParquetUtils.withHadoopReadOptions(new
HoodieAvroParquetReaderBuilder<IndexedRecord>(inputFile)
+
.withTableSchema(getAvroSchemaConverter(hadoopConf).convert(readSchema)),
inputFile)
.set(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS,
hadoopConf.get(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS))
.set(ParquetInputFormat.STRICT_TYPE_CHECKING,
hadoopConf.get(ParquetInputFormat.STRICT_TYPE_CHECKING))
.build();
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/parquet/avro/HoodieAvroReadSupport.java
b/hudi-hadoop-common/src/main/java/org/apache/parquet/avro/HoodieAvroReadSupport.java
index a895db255639..fe7532ff780a 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/parquet/avro/HoodieAvroReadSupport.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/parquet/avro/HoodieAvroReadSupport.java
@@ -88,7 +88,7 @@ public class HoodieAvroReadSupport<T> extends
AvroReadSupport<T> {
String requestedProjectionString =
configuration.get(AVRO_REQUESTED_PROJECTION);
if (requestedProjectionString != null) {
HoodieSchema avroRequestedProjection =
HoodieSchema.parse(requestedProjectionString);
- Configuration conf = new Configuration();
+ Configuration conf = new Configuration(false);
configuration.forEach(entry -> conf.set(entry.getKey(),
entry.getValue()));
projection = new
AvroSchemaConverter(conf).convert(avroRequestedProjection.toAvroSchema());
}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/FooterKeyDecryptionFactory.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/FooterKeyDecryptionFactory.java
new file mode 100644
index 000000000000..6863dff14936
--- /dev/null
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/FooterKeyDecryptionFactory.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.testutils;
+
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.avro.AvroParquetWriter;
+import org.apache.parquet.crypto.DecryptionPropertiesFactory;
+import org.apache.parquet.crypto.FileDecryptionProperties;
+import org.apache.parquet.crypto.FileEncryptionProperties;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.util.HadoopOutputFile;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * A parquet decryption factory ({@code parquet.crypto.factory.class}) that
decrypts files written by
+ * {@link #writeEncryptedFile} with one fixed footer key, and records the file
path of every request.
+ */
+public class FooterKeyDecryptionFactory implements DecryptionPropertiesFactory
{
+
+ public static final String CRYPTO_FACTORY_CLASS =
"parquet.crypto.factory.class";
+ public static final Schema SCHEMA = SchemaBuilder.record("record").fields()
+ .requiredString("_row_key")
+ .requiredLong("ts")
+ .endRecord();
+
+ private static final byte[] FOOTER_KEY =
"0123456789012345".getBytes(StandardCharsets.UTF_8);
+ private static final List<Path> REQUESTED_PATHS =
Collections.synchronizedList(new ArrayList<>());
+
+ @Override
+ public FileDecryptionProperties getFileDecryptionProperties(Configuration
hadoopConfig, Path filePath) {
+ REQUESTED_PATHS.add(filePath);
+ return
FileDecryptionProperties.builder().withFooterKey(FOOTER_KEY).build();
+ }
+
+ /**
+ * Returns and clears the file paths requested so far.
+ */
+ public static List<Path> drainRequestedPaths() {
+ synchronized (REQUESTED_PATHS) {
+ List<Path> paths = new ArrayList<>(REQUESTED_PATHS);
+ REQUESTED_PATHS.clear();
+ return paths;
+ }
+ }
+
+ /**
+ * Writes {@code numRecords} records of {@link #SCHEMA} with an encrypted
footer and encrypted columns.
+ */
+ public static void writeEncryptedFile(Configuration conf, Path path, int
numRecords) throws IOException {
+ writeFile(conf, path, numRecords, true);
+ }
+
+ /**
+ * Writes {@code numRecords} records of {@link #SCHEMA}, encrypted as by
{@link #writeEncryptedFile} or in plaintext.
+ */
+ public static void writeFile(Configuration conf, Path path, int numRecords,
boolean encrypted) throws IOException {
+ try (ParquetWriter<GenericRecord> writer =
AvroParquetWriter.<GenericRecord>builder(HadoopOutputFile.fromPath(path, conf))
+ .withSchema(SCHEMA)
+ .withConf(conf)
+ .withEncryption(encrypted ?
FileEncryptionProperties.builder(FOOTER_KEY).build() : null)
+ .build()) {
+ for (int i = 0; i < numRecords; i++) {
+ GenericRecord record = new GenericData.Record(SCHEMA);
+ record.put("_row_key", "key" + i);
+ record.put("ts", (long) i);
+ writer.write(record);
+ }
+ }
+ }
+}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HadoopDefaultResourceLoadCounter.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HadoopDefaultResourceLoadCounter.java
new file mode 100644
index 000000000000..693f4ad6c30b
--- /dev/null
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HadoopDefaultResourceLoadCounter.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.testutils;
+
+import org.junit.jupiter.api.function.Executable;
+
+import java.net.URL;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Counts the Hadoop default resources ({@code core-default.xml}, {@code
core-site.xml}) loaded
+ * while an action runs on the current thread.
+ *
+ * <p>A Hadoop {@code Configuration} captures the thread context class loader
when it is created
+ * and loads the default resources through it on first access, while a copy of
an existing
+ * configuration keeps the loader of its source. Running the action under a
counting context
+ * class loader therefore counts only the configurations the action creates
from scratch.
+ */
+public final class HadoopDefaultResourceLoadCounter {
+
+ private HadoopDefaultResourceLoadCounter() {
+ }
+
+ public static int countDefaultResourceLoads(Executable action) throws
Throwable {
+ Thread thread = Thread.currentThread();
+ ClassLoader original = thread.getContextClassLoader();
+ CountingClassLoader counting = new CountingClassLoader(original);
+ thread.setContextClassLoader(counting);
+ try {
+ action.execute();
+ } finally {
+ thread.setContextClassLoader(original);
+ }
+ return counting.loads.get();
+ }
+
+ private static class CountingClassLoader extends ClassLoader {
+ private final AtomicInteger loads = new AtomicInteger();
+
+ CountingClassLoader(ClassLoader parent) {
+ super(parent);
+ }
+
+ @Override
+ public URL getResource(String name) {
+ if ("core-default.xml".equals(name) || "core-site.xml".equals(name)) {
+ loads.incrementAndGet();
+ }
+ return super.getResource(name);
+ }
+ }
+}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieAvroParquetReader.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieAvroParquetReader.java
new file mode 100644
index 000000000000..6d3b2c26d5a2
--- /dev/null
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieAvroParquetReader.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage.hadoop;
+
+import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.engine.LocalTaskContextSupplier;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.testutils.FooterKeyDecryptionFactory;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static
org.apache.hudi.common.testutils.HadoopDefaultResourceLoadCounter.countDefaultResourceLoads;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/**
+ * Tests for {@link HoodieAvroParquetReader}.
+ */
+class TestHoodieAvroParquetReader {
+
+ @TempDir
+ java.nio.file.Path tmpDir;
+
+ @Test
+ void testReadDoesNotLoadHadoopDefaultResources() throws Throwable {
+ HoodieStorage storage = HoodieTestUtils.getStorage(tmpDir.toString());
+ StoragePath parquetPath = new
StoragePath(tmpDir.resolve("file.parquet").toAbsolutePath().toString());
+ List<GenericRecord> records = new
HoodieTestDataGenerator(0xDEED).generateGenericRecords(20);
+ HoodieSchema schema =
HoodieSchema.fromAvroSchema(records.get(0).getSchema());
+ HoodieConfig config = new HoodieConfig();
+ config.setValue(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME,
"uncompressed");
+ HoodieAvroParquetWriter writer = (HoodieAvroParquetWriter) new
HoodieAvroFileWriterFactory(storage)
+ .newParquetFileWriter("001", parquetPath, config, schema, new
LocalTaskContextSupplier());
+ for (GenericRecord record : records) {
+ writer.writeAvro((String) record.get("_row_key"), record);
+ }
+ writer.close();
+
+ List<String> readKeys = new ArrayList<>();
+ int loads = countDefaultResourceLoads(() -> {
+ try (HoodieAvroParquetReader reader = new
HoodieAvroParquetReader(storage, parquetPath);
+ ClosableIterator<IndexedRecord> iterator =
reader.getIndexedRecordIterator(schema)) {
+ iterator.forEachRemaining(record -> readKeys.add(((GenericRecord)
record).get("_row_key").toString()));
+ }
+ });
+ assertEquals(records.stream().map(record ->
record.get("_row_key").toString()).collect(Collectors.toList()), readKeys);
+ assertEquals(0, loads, "Reading a parquet file must reuse the storage
configuration");
+ }
+
+ @Test
+ void testReadEncryptedFileResolvesDecryptionWithFilePath() throws Exception {
+ HoodieStorage storage = HoodieTestUtils.getStorage(tmpDir.toString());
+ storage.getConf().set(FooterKeyDecryptionFactory.CRYPTO_FACTORY_CLASS,
FooterKeyDecryptionFactory.class.getName());
+ StoragePath parquetPath = new
StoragePath(tmpDir.resolve("encrypted.parquet").toAbsolutePath().toString());
+
FooterKeyDecryptionFactory.writeEncryptedFile(storage.getConf().unwrapAs(Configuration.class),
new Path(parquetPath.toUri()), 10);
+ FooterKeyDecryptionFactory.drainRequestedPaths();
+
+ List<String> readKeys = new ArrayList<>();
+ try (HoodieAvroParquetReader reader = new HoodieAvroParquetReader(storage,
parquetPath);
+ ClosableIterator<IndexedRecord> iterator =
reader.getIndexedRecordIterator(HoodieSchema.fromAvroSchema(FooterKeyDecryptionFactory.SCHEMA)))
{
+ iterator.forEachRemaining(record -> readKeys.add(((GenericRecord)
record).get("_row_key").toString()));
+ }
+ assertEquals(IntStream.range(0, 10).mapToObj(i -> "key" +
i).collect(Collectors.toList()), readKeys);
+ List<Path> requestedPaths =
FooterKeyDecryptionFactory.drainRequestedPaths();
+ assertFalse(requestedPaths.isEmpty());
+ requestedPaths.forEach(requested -> assertEquals("encrypted.parquet",
requested == null ? null : requested.getName()));
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
index 9df771b115e6..b9470141aac1 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
@@ -36,6 +36,9 @@ import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
+import org.apache.hadoop.mapred.JobConf
+import org.apache.parquet.filter2.predicate.FilterPredicate
+import org.apache.parquet.hadoop.ParquetInputFormat
import org.apache.parquet.hadoop.metadata.FileMetaData
import org.apache.parquet.schema.{GroupType, MessageType, Type => ParquetType}
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
@@ -50,7 +53,7 @@ import java.time.ZoneId
import scala.collection.convert.ImplicitConversions.`collection
AsScalaIterable`
-class ParquetSchemaEvolutionUtils(sharedConf: Configuration,
+class ParquetSchemaEvolutionUtils(readConf: Configuration,
filePath: Path,
requiredSchema: StructType,
partitionSchema: StructType,
@@ -62,14 +65,14 @@ class ParquetSchemaEvolutionUtils(sharedConf: Configuration,
private lazy val schemaUtils: HoodieSchemaUtils = sparkAdapter.getSchemaUtils
- private lazy val tablePath: String =
sharedConf.get(SparkInternalSchemaConverter.HOODIE_TABLE_PATH)
+ private lazy val tablePath: String =
readConf.get(SparkInternalSchemaConverter.HOODIE_TABLE_PATH)
private lazy val fileSchema: InternalSchema = if (shouldUseInternalSchema) {
val commitInstantTime = FSUtils.getCommitTime(filePath.getName).toLong
//TODO: HARDCODED TIMELINE OBJECT
- val validCommits =
sharedConf.get(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST)
+ val validCommits =
readConf.get(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST)
val layout =
TimelineLayout.fromVersion(TimelineLayoutVersion.CURR_LAYOUT_VERSION)
InternalSchemaCache.getInternalSchemaByVersionId(commitInstantTime,
tablePath,
- HoodieStorageUtils.getStorage(tablePath,
HadoopFSUtils.getStorageConf(sharedConf)), if (validCommits == null) "" else
validCommits, layout)
+ HoodieStorageUtils.getStorage(tablePath,
HadoopFSUtils.getStorageConf(readConf)), if (validCommits == null) "" else
validCommits, layout)
} else {
null
}
@@ -80,9 +83,26 @@ 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 or the
pushed row-group filter. The
+ * read configuration is never modified, since other readers may share it,
so callers must not modify the
+ * returned configuration either.
+ */
+ def getFileReadConf(footerFileMetaData: FileMetaData,
+ enableVectorizedReader: Boolean,
+ pushedFilter: Option[FilterPredicate]): Configuration = {
+ var fileReadConf: Configuration = readConf
+ def fileOwnConf(): Configuration = {
+ if (fileReadConf eq readConf) {
+ // A JobConf, so the task attempt context built on it does not copy it
again: JobContextImpl reuses a
+ // JobConf and copies any other Configuration
+ fileReadConf = new JobConf(readConf)
+ }
+ fileReadConf
+ }
+ def setRequestedSchema(schema: StructType): Unit =
+ fileOwnConf().set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
schema.json)
typeChangeInfos = if (shouldUseInternalSchema) {
// Empty projections (count(*), select 1) read no column data, so there
is nothing to
// reconstruct - and querySchemaOption is the UNPRUNED table schema in
that case (see
@@ -93,14 +113,14 @@ class ParquetSchemaEvolutionUtils(sharedConf:
Configuration,
val mergedInternalSchema = new InternalSchemaMerger(fileSchema,
querySchemaOption.get(), true, true).mergeSchema()
val mergedSchema =
SparkInternalSchemaConverter.constructSparkSchemaFromInternalSchema(mergedInternalSchema)
- hadoopAttemptConf.set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
mergedSchema.json)
+ setRequestedSchema(mergedSchema)
SparkInternalSchemaConverter.collectTypeChangedCols(querySchemaOption.get(),
mergedInternalSchema)
} else {
- val (implicitTypeChangeInfo, sparkRequestSchema) =
HoodieParquetFileFormatHelper.buildImplicitSchemaChangeInfo(hadoopAttemptConf,
footerFileMetaData, requiredSchema)
+ val (implicitTypeChangeInfo, sparkRequestSchema) =
HoodieParquetFileFormatHelper.buildImplicitSchemaChangeInfo(readConf,
footerFileMetaData, requiredSchema)
if (!implicitTypeChangeInfo.isEmpty) {
shouldUseInternalSchema = true
- hadoopAttemptConf.set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
sparkRequestSchema.json)
+ setRequestedSchema(sparkRequestSchema)
}
implicitTypeChangeInfo
}
@@ -112,7 +132,9 @@ class ParquetSchemaEvolutionUtils(sharedConf: Configuration,
"To workaround this issue, set
spark.sql.parquet.enableVectorizedReader=false.")
}
- hadoopAttemptConf
+ // Row-group level push-down, not individual records
+ pushedFilter.foreach(ParquetInputFormat.setFilterPredicate(fileOwnConf(),
_))
+ fileReadConf
}
def generateUnsafeProjection(fullSchema: Seq[AttributeReference],
timeZoneId: Option[String]): UnsafeProjection = {
@@ -255,7 +277,7 @@ object ParquetSchemaEvolutionUtils {
* names both routes rather than blaming pushVariantIntoScan on a read that
never set it.
* Real support is #18285.
*
- * Shared by [[ParquetSchemaEvolutionUtils.getHadoopConfClone]] and the
per-version legacy
+ * Shared by [[ParquetSchemaEvolutionUtils.getFileReadConf]] and the
per-version legacy
* file formats, which carry a copy of the same schema-merge block. Callers
gate on a
* non-empty projection: empty-projection queries (count(*), select 1) read
no column data
* and must keep working, and the query schema is unpruned in that case.
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkParquetReaderBase.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkParquetReaderBase.scala
index d901c42dae6e..bdc4ed9700d6 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkParquetReaderBase.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkParquetReaderBase.scala
@@ -25,6 +25,7 @@ import org.apache.hudi.common.util
import org.apache.hudi.storage.StorageConfiguration
import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.mapred.JobConf
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.execution.datasources.{PartitionedFile,
SparkColumnarFileReader}
import org.apache.spark.sql.internal.SQLConf
@@ -64,7 +65,9 @@ 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:
+ // JobContextImpl reuses a JobConf and copies any other Configuration
+ val conf = new JobConf(storageConf.unwrap())
conf.set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
requiredSchema.json)
conf.set(ParquetWriteSupport.SPARK_ROW_SCHEMA, requiredSchema.json)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
index 0df5f28af986..463c1993b888 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
@@ -22,14 +22,21 @@ package org.apache.spark.sql.execution.datasources.parquet
import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaField,
HoodieSchemaType}
import org.apache.hudi.common.schema.internal.InternalSchema
import org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter
+import org.apache.hudi.common.util
import org.apache.hudi.exception.HoodieException
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.filter2.compat.FilterCompat
+import org.apache.parquet.filter2.predicate.{FilterApi, FilterPredicate}
+import org.apache.parquet.hadoop.ParquetInputFormat
import org.apache.parquet.hadoop.metadata.FileMetaData
import org.apache.parquet.schema.{MessageType, Type, Types}
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import
org.apache.spark.sql.execution.datasources.parquet.VariantParquetTestFixtures.{shreddedVariant,
stringKeyMap, threeLevelList, twoLevelList, unshreddedVariant}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources.{AlwaysFalse, AlwaysTrue, And,
EqualNullSafe, EqualTo, Filter, GreaterThan, GreaterThanOrEqual, In, IsNotNull,
IsNull, LessThan, LessThanOrEqual, Not, Or, StringContains, StringEndsWith,
StringStartsWith}
-import org.apache.spark.sql.types.{ArrayType, BinaryType, IntegerType,
MapType, MetadataBuilder, StringType, StructField, StructType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, IntegerType,
LongType, MapType, MetadataBuilder, StringType, StructField, StructType}
import org.junit.jupiter.api.{Assertions, Test}
import java.util.{Arrays, Collections, HashMap}
@@ -387,6 +394,48 @@ class TestParquetSchemaEvolutionUtils {
ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(untouched, null,
null))
}
+ /**
+ * The read configuration may be shared by other readers: a file that needs
its own requested schema or a
+ * pushed filter gets a copy, and any other file reads with the read
configuration itself.
+ */
+ @Test
+ def testGetFileReadConfNeverModifiesTheReadConf(): Unit = {
+ val readConf = new Configuration(false)
+ // The flags ParquetToSparkSchemaConverter needs, as
SparkParquetReaderBase.read sets them
+ readConf.setBoolean(SQLConf.CASE_SENSITIVE.key, false)
+ readConf.setBoolean(SQLConf.PARQUET_BINARY_AS_STRING.key, false)
+ readConf.setBoolean(SQLConf.PARQUET_INT96_AS_TIMESTAMP.key, true)
+ readConf.setBoolean(SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key, false)
+ readConf.setBoolean("spark.sql.parquet.inferTimestampNTZ.enabled", true)
+ val requiredSchema = new StructType().add("id", LongType)
+ readConf.set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
requiredSchema.json)
+
+ def fileReadConf(fileType: PrimitiveTypeName, pushedFilter:
Option[FilterPredicate]): Configuration =
+ new ParquetSchemaEvolutionUtils(readConf, new Path("/tmp/file.parquet"),
requiredSchema, new StructType(), util.Option.empty())
+ .getFileReadConf(footerOf(Types.optional(fileType).named("id")),
enableVectorizedReader = false, pushedFilter)
+
+ def pushedFilterOf(conf: Configuration): FilterPredicate =
ParquetInputFormat.getFilter(conf) match {
+ case predicate: FilterCompat.FilterPredicateCompat =>
predicate.getFilterPredicate
+ case _ => null
+ }
+
+ val filter = FilterApi.eq(FilterApi.longColumn("id"),
java.lang.Long.valueOf(1L))
+ Assertions.assertSame(readConf, fileReadConf(PrimitiveTypeName.INT64,
None))
+ val filtered = fileReadConf(PrimitiveTypeName.INT64, Some(filter))
+ Assertions.assertNotSame(readConf, filtered)
+ Assertions.assertEquals(requiredSchema.json,
filtered.get(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA))
+ Assertions.assertEquals(filter, pushedFilterOf(filtered))
+ // An int file read as long gets a copy that requests the int column
+ for (pushedFilter <- Seq(None, Some(filter))) {
+ val promoted = fileReadConf(PrimitiveTypeName.INT32, pushedFilter)
+ Assertions.assertNotSame(readConf, promoted)
+ Assertions.assertEquals(new StructType().add("id", IntegerType).json,
promoted.get(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA))
+ Assertions.assertEquals(pushedFilter.orNull, pushedFilterOf(promoted))
+ }
+ Assertions.assertEquals(requiredSchema.json,
readConf.get(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA))
+ Assertions.assertNull(pushedFilterOf(readConf))
+ }
+
/** An internal schema over the given top-level columns, in order; field ids
are positional. */
private def internalSchemaOf(fields: (String, HoodieSchema)*):
InternalSchema =
InternalSchemaConverter.convert(HoodieSchema.createRecord("query",
"org.apache.hudi.test", null,
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/TestHoodieSparkParquetReader.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/TestHoodieSparkParquetReader.java
new file mode 100644
index 000000000000..37f12b0e91a5
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/TestHoodieSparkParquetReader.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.engine.LocalTaskContextSupplier;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.testutils.FooterKeyDecryptionFactory;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.core.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.io.storage.hadoop.HoodieAvroParquetWriter;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.sources.EqualTo;
+import org.apache.spark.sql.sources.Filter;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import static
org.apache.hudi.common.testutils.HadoopDefaultResourceLoadCounter.countDefaultResourceLoads;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/**
+ * Tests for {@link HoodieSparkParquetReader}.
+ */
+class TestHoodieSparkParquetReader {
+
+ private static final int NUM_RECORDS = 20;
+
+ @TempDir
+ java.nio.file.Path tmpDir;
+
+ private HoodieStorage storage;
+ private StoragePath parquetPath;
+ private HoodieSchema schema;
+ private List<GenericRecord> records;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ storage = HoodieTestUtils.getStorage(tmpDir.toString());
+ parquetPath = new
StoragePath(tmpDir.resolve("file.parquet").toAbsolutePath().toString());
+ records = new
HoodieTestDataGenerator(0xDEED).generateGenericRecords(NUM_RECORDS);
+ schema = HoodieSchema.fromAvroSchema(records.get(0).getSchema());
+ HoodieConfig config = new HoodieConfig();
+ config.setValue(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME,
"uncompressed");
+ HoodieAvroParquetWriter writer = (HoodieAvroParquetWriter)
HoodieFileWriterFactory.getFileWriter(
+ "001", parquetPath, storage, config, schema, new
LocalTaskContextSupplier(), HoodieRecord.HoodieRecordType.AVRO);
+ for (GenericRecord record : records) {
+ writer.writeAvro((String) record.get("_row_key"), record);
+ }
+ writer.close();
+ }
+
+ @Test
+ void testReadDoesNotLoadHadoopDefaultResources() throws Throwable {
+ int[] rowCount = new int[1];
+ int loads = countDefaultResourceLoads(() -> rowCount[0] =
readRows(Collections.emptyList()));
+ assertEquals(NUM_RECORDS, rowCount[0]);
+ assertEquals(0, loads, "Reading a parquet file must reuse the storage
configuration");
+ }
+
+ @Test
+ void testReadAppliesPushedDownFilters() throws Exception {
+ String firstKey = (String) records.get(0).get("_row_key");
+ // Row group statistics keep the matching row group, and drop it for a key
above every UUID key.
+ assertEquals(NUM_RECORDS, readRows(Collections.singletonList(new
EqualTo("_row_key", firstKey))));
+ assertEquals(0, readRows(Collections.singletonList(new EqualTo("_row_key",
"~"))));
+ }
+
+ @Test
+ void testReadEncryptedFileResolvesDecryptionWithFilePath() throws Exception {
+ storage.getConf().set(FooterKeyDecryptionFactory.CRYPTO_FACTORY_CLASS,
FooterKeyDecryptionFactory.class.getName());
+ StoragePath encryptedPath = new
StoragePath(tmpDir.resolve("encrypted.parquet").toAbsolutePath().toString());
+
FooterKeyDecryptionFactory.writeEncryptedFile(storage.getConf().unwrapAs(Configuration.class),
new Path(encryptedPath.toUri()), 10);
+ FooterKeyDecryptionFactory.drainRequestedPaths();
+
+ List<Long> readTs = new ArrayList<>();
+ try (HoodieSparkParquetReader reader = (HoodieSparkParquetReader) new
HoodieSparkFileReaderFactory(storage)
+ .newParquetFileReader(encryptedPath);
+ ClosableIterator<UnsafeRow> iterator =
reader.getUnsafeRowIterator(HoodieSchema.fromAvroSchema(FooterKeyDecryptionFactory.SCHEMA)))
{
+ iterator.forEachRemaining(row -> readTs.add(row.getLong(1)));
+ }
+ assertEquals(LongStream.range(0, 10).boxed().collect(Collectors.toList()),
readTs);
+ List<Path> requestedPaths =
FooterKeyDecryptionFactory.drainRequestedPaths();
+ assertFalse(requestedPaths.isEmpty());
+ requestedPaths.forEach(requested -> assertEquals("encrypted.parquet",
requested == null ? null : requested.getName()));
+ }
+
+ private int readRows(List<Filter> filters) throws Exception {
+ int count = 0;
+ try (HoodieSparkParquetReader reader = (HoodieSparkParquetReader) new
HoodieSparkFileReaderFactory(storage)
+ .newParquetFileReader(parquetPath);
+ ClosableIterator<UnsafeRow> iterator =
reader.getUnsafeRowIterator(schema, filters)) {
+ while (iterator.hasNext()) {
+ iterator.next();
+ count++;
+ }
+ }
+ return count;
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/TestParquetReadOptionsParity.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/TestParquetReadOptionsParity.java
new file mode 100644
index 000000000000..84076f7dfbe9
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/TestParquetReadOptionsParity.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import org.apache.hudi.common.testutils.FooterKeyDecryptionFactory;
+import org.apache.hudi.common.util.ParquetUtils;
+import org.apache.hudi.storage.StoragePath;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.HadoopReadOptions;
+import org.apache.parquet.ParquetReadOptions;
+import org.apache.parquet.avro.HoodieAvroParquetReaderBuilder;
+import org.apache.parquet.crypto.FileDecryptionProperties;
+import org.apache.parquet.filter2.compat.FilterCompat;
+import org.apache.parquet.filter2.predicate.FilterApi;
+import org.apache.parquet.hadoop.ParquetInputFormat;
+import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.util.HadoopInputFile;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+/**
+ * Tests {@link ParquetUtils#withHadoopReadOptions} against the parquet
version on the classpath. It lives in
+ * this module, rather than next to {@link ParquetUtils}, because the parquet
version differs per Spark
+ * profile and CI runs this module's tests under each one.
+ */
+class TestParquetReadOptionsParity {
+
+ @TempDir
+ java.nio.file.Path tempDir;
+
+ /**
+ * A reader built with {@link ParquetUtils#withHadoopReadOptions} must get
the read options parquet builds
+ * from the configuration and the file path, field by field, and read the
same rows as a reader built from
+ * the path with {@code withConf}, both with the read options left at their
defaults and with them set.
+ */
+ @ParameterizedTest
+ @CsvSource({"false,false", "false,true", "true,false", "true,true"})
+ void testWithHadoopReadOptionsMatchesPathBasedReader(boolean encrypted,
boolean setReadOptions) throws Exception {
+ Configuration conf = new Configuration();
+ Path path = new Path(tempDir.resolve("options.parquet").toUri());
+ FooterKeyDecryptionFactory.writeFile(conf, path, 10, encrypted);
+ if (encrypted) {
+ conf.set(FooterKeyDecryptionFactory.CRYPTO_FACTORY_CLASS,
FooterKeyDecryptionFactory.class.getName());
+ }
+ if (setReadOptions) {
+ // Non-default values, so options that ignore the configuration differ
+ conf.setBoolean(ParquetInputFormat.STATS_FILTERING_ENABLED, false);
+ conf.setBoolean(ParquetInputFormat.DICTIONARY_FILTERING_ENABLED, false);
+ conf.setBoolean(ParquetInputFormat.COLUMN_INDEX_FILTERING_ENABLED,
false);
+ conf.setBoolean(ParquetInputFormat.BLOOM_FILTERING_ENABLED, false);
+ conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, true);
+ conf.setInt("parquet.read.allocation.size", 4 * 1024 * 1024);
+ ParquetInputFormat.setFilterPredicate(conf,
FilterApi.eq(FilterApi.longColumn("ts"), 5L));
+ }
+
+ HadoopInputFile inputFile = HadoopInputFile.fromPath(path, conf);
+ ParquetReadOptions expectedOptions = HadoopReadOptions.builder(conf,
inputFile.getPath()).build();
+ FooterKeyDecryptionFactory.drainRequestedPaths();
+ try (ParquetReader<IndexedRecord> reader =
ParquetUtils.withHadoopReadOptions(
+ new HoodieAvroParquetReaderBuilder<IndexedRecord>(inputFile),
inputFile).build()) {
+ assertEquals(encrypted ? Collections.singletonList(inputFile.getPath())
: Collections.emptyList(),
+ FooterKeyDecryptionFactory.drainRequestedPaths());
+ Field options = ParquetReader.class.getDeclaredField("options");
+ options.setAccessible(true);
+ assertSameFields(expectedOptions, options.get(reader), "options");
+
+ List<String> keys = readKeys(reader);
+ assertEquals(setReadOptions ? Collections.singletonList("key5")
+ : IntStream.range(0, 10).mapToObj(i -> "key" +
i).collect(Collectors.toList()), keys);
+ try (ParquetReader<IndexedRecord> pathReader = new
HoodieAvroParquetReaderBuilder<IndexedRecord>(
+ new StoragePath(path.toUri())).withConf(conf).build()) {
+ assertEquals(readKeys(pathReader), keys);
+ }
+ }
+ }
+
+ private static void assertSameFields(Object expected, Object actual, String
name) throws ReflectiveOperationException {
+ assertEquals(expected.getClass(), actual.getClass(), name);
+ for (Class<?> clazz = expected.getClass(); clazz != Object.class; clazz =
clazz.getSuperclass()) {
+ for (Field field : clazz.getDeclaredFields()) {
+ if (Modifier.isStatic(field.getModifiers())) {
+ continue;
+ }
+ field.setAccessible(true);
+ Object expectedValue = field.get(expected);
+ Object actualValue = field.get(actual);
+ String fieldName = name + "." + field.getName();
+ if (expectedValue == null || actualValue == null || expectedValue
instanceof Configuration) {
+ assertSame(expectedValue, actualValue, fieldName);
+ } else if (expectedValue instanceof byte[]) {
+ assertArrayEquals((byte[]) expectedValue, (byte[]) actualValue,
fieldName);
+ } else if (expectedValue instanceof FileDecryptionProperties) {
+ assertSameFields(expectedValue, actualValue, fieldName);
+ } else if (expectedValue instanceof
FilterCompat.FilterPredicateCompat) {
+ assertEquals(((FilterCompat.FilterPredicateCompat)
expectedValue).getFilterPredicate(),
+ ((FilterCompat.FilterPredicateCompat)
actualValue).getFilterPredicate(), fieldName);
+ } else if (expectedValue.getClass().getMethod("equals",
Object.class).getDeclaringClass() != Object.class) {
+ assertEquals(expectedValue, actualValue, fieldName);
+ } else {
+ // Created per build without value equality (codec factory,
allocator, configuration wrapper)
+ assertEquals(expectedValue.getClass(), actualValue.getClass(),
fieldName);
+ }
+ }
+ }
+ }
+
+ private static List<String> readKeys(ParquetReader<IndexedRecord> reader)
throws IOException {
+ List<String> keys = new ArrayList<>();
+ for (IndexedRecord record = reader.read(); record != null; record =
reader.read()) {
+ keys.add(((GenericRecord) record).get("_row_key").toString());
+ }
+ return keys;
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBasicSchemaEvolution.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBasicSchemaEvolution.scala
index 07829da4e0e9..709c2727dd16 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBasicSchemaEvolution.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBasicSchemaEvolution.scala
@@ -468,7 +468,7 @@ class TestBasicSchemaEvolution extends
HoodieSparkClientTestBase with ScalaAsser
* spans a base-only int file (p2) and an int base file merged with a long
log file (p1). The
* top-level `age` promotion is atomic and is read with the vectorized
parquet reader on. When the
* same promotion is applied inside the `nested` struct the changed
top-level column is no longer
- * atomic: ParquetSchemaEvolutionUtils.getHadoopConfClone must reject it
fast on the base slice
+ * atomic: ParquetSchemaEvolutionUtils.getFileReadConf must reject it fast
on the base slice
* instead of returning corrupt columns, and the workaround it advertises
(disabling the vectorized
* reader) must actually widen `nested.a` across both shapes. COW is covered
by
* TestLegacyParquetReadPath#testCowSnapshotReadWithNestedTypeChange.
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/io/storage/ParquetPerFileReadBenchmark.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/io/storage/ParquetPerFileReadBenchmark.scala
new file mode 100644
index 000000000000..504ef0e7c4b0
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/io/storage/ParquetPerFileReadBenchmark.scala
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage
+
+import org.apache.hudi.SparkAdapterSupport.sparkAdapter
+import org.apache.hudi.common.util.{HoodieStorageUtils, Option => HOption}
+import org.apache.hudi.hadoop.fs.HadoopFSUtils
+import org.apache.hudi.storage.StoragePath
+import org.apache.hudi.util.CloseableInternalRowIterator
+
+import org.apache.hadoop.conf.Configuration
+import
org.apache.parquet.avro.HoodieAvroParquetSchemaConverter.getAvroSchemaConverter
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.avro.HoodieSparkSchemaConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.types.StructType
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty
+import org.junit.jupiter.api.io.TempDir
+
+import java.lang.management.ManagementFactory
+import java.nio.file.{Files, Path}
+
+import scala.collection.JavaConverters._
+
+/**
+ * Manual benchmark of the per-file CPU cost of the Spark parquet read paths:
the log-file reader
+ * ([[HoodieSparkParquetReader]]) and the base-file reader built by the Spark
adapter. Runs only when
+ * `benchmark.files` is set, so it stays out of CI. Every round reads the same
small files and checks
+ * the row count and a checksum, so a faster but wrong reader fails the run.
Run it with surefire
+ * directly, since `mvn test` in this module also runs every scalatest suite:
+ *
+ * {{{
+ * mvn test-compile -pl hudi-spark-datasource/hudi-spark
+ * mvn surefire:test -pl hudi-spark-datasource/hudi-spark
-Dtest=ParquetPerFileReadBenchmark \
+ * -Dbenchmark.files=400 -Dbenchmark.rounds=10 -Dbenchmark.warmupRounds=3
+ * }}}
+ */
+@EnabledIfSystemProperty(named = "benchmark.files", matches = "\\d+")
+class ParquetPerFileReadBenchmark {
+
+ @TempDir
+ var tmpDir: Path = _
+
+ @Test
+ def benchmark(): Unit = {
+ val numFiles = Integer.getInteger("benchmark.files").intValue()
+ val rounds = Integer.getInteger("benchmark.rounds", 10).intValue()
+ val warmupRounds = Integer.getInteger("benchmark.warmupRounds",
3).intValue()
+ val rowsPerFile = 20
+ val spark = SparkSession.builder()
+ .master("local[1]")
+ .appName("ParquetPerFileReadBenchmark")
+ .config("spark.ui.enabled", "false")
+ .getOrCreate()
+ try {
+ val dir = tmpDir.resolve("data").toString
+ spark.range(0, numFiles.toLong * rowsPerFile, 1, numFiles)
+ .selectExpr("concat('key', cast(id as string)) as key", "id as ts",
"concat('value', cast(id as string)) as value")
+ .write.parquet(dir)
+ val files = Files.list(tmpDir.resolve("data")).iterator().asScala
+
.filter(_.getFileName.toString.endsWith(".parquet")).map(_.toString).toSeq.sorted
+ assertEquals(numFiles, files.size)
+
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val storageConf = HadoopFSUtils.getStorageConf(hadoopConf)
+ val storage = HoodieStorageUtils.getStorage(new StoragePath(files.head),
storageConf)
+ val structType = spark.read.parquet(files.head).schema
+ val schema = HoodieSparkSchemaConverters.toHoodieType(structType,
nullable = false, "record", "")
+ val baseFileReader = sparkAdapter.createParquetFileReader(vectorized =
true, spark.sessionState.conf,
+ Map(FileFormat.OPTION_RETURNING_BATCH -> "false"), hadoopConf)
+ val expectedChecksum = (0L until numFiles.toLong * rowsPerFile).sum
+
+ def readLogPath(file: String): (Long, Long) = {
+ val reader = new
HoodieSparkFileReaderFactory(storage).newParquetFileReader(new
StoragePath(file))
+ .asInstanceOf[HoodieSparkParquetReader]
+ try {
+ val iterator = reader.getUnsafeRowIterator(schema)
+ var rows = 0L
+ var checksum = 0L
+ while (iterator.hasNext) {
+ checksum += iterator.next().getLong(1)
+ rows += 1
+ }
+ (rows, checksum)
+ } finally {
+ reader.close()
+ }
+ }
+
+ def readBasePath(file: String): (Long, Long) = {
+ val partitionedFile = sparkAdapter.getSparkPartitionedFileUtils
+ .createPartitionedFile(InternalRow.empty, new StoragePath(file), 0,
Files.size(java.nio.file.Paths.get(file)))
+ // Mirrors SparkFileFormatInternalRowReaderContext, which converts the
table schema for every base file.
+ val tableSchema =
getAvroSchemaConverter(storageConf.unwrap()).convert(schema)
+ val iterator = new
CloseableInternalRowIterator(baseFileReader.read(partitionedFile, structType,
+ new StructType(), HOption.empty(), Seq.empty, storageConf,
HOption.of(tableSchema)))
+ try {
+ var rows = 0L
+ var checksum = 0L
+ while (iterator.hasNext) {
+ checksum += iterator.next().getLong(1)
+ rows += 1
+ }
+ (rows, checksum)
+ } finally {
+ iterator.close()
+ }
+ }
+
+ // Reference costs of single per-file steps, measured the same way; they
read no rows.
+ def copyConf(file: String): (Long, Long) = {
+ new Configuration(hadoopConf)
+ (-1L, -1L)
+ }
+
+ def newConfiguration(file: String): (Long, Long) = {
+ new Configuration().size()
+ (-1L, -1L)
+ }
+
+ def convertTableSchema(file: String): (Long, Long) = {
+ getAvroSchemaConverter(storageConf.unwrap()).convert(schema)
+ (-1L, -1L)
+ }
+
+ val threadMx = ManagementFactory.getThreadMXBean
+ val paths = Seq("log" -> (readLogPath _), "base" -> (readBasePath _),
"confCopy" -> (copyConf _),
+ "newConfiguration" -> (newConfiguration _), "tableSchemaConversion" ->
(convertTableSchema _))
+ val results = paths.map { case (name, read) =>
+ val perFileCpuMicros = (0 until warmupRounds + rounds).map { _ =>
+ val start = threadMx.getCurrentThreadCpuTime
+ var rows = 0L
+ var checksum = 0L
+ files.foreach { file =>
+ val (r, c) = read(file)
+ rows += r
+ checksum += c
+ }
+ val elapsed = threadMx.getCurrentThreadCpuTime - start
+ if (rows >= 0) {
+ assertEquals(numFiles.toLong * rowsPerFile, rows)
+ assertEquals(expectedChecksum, checksum)
+ }
+ elapsed / 1000.0 / numFiles
+ }.drop(warmupRounds).sorted
+ name -> perFileCpuMicros
+ }
+ // scalastyle:off println
+ println(s"ParquetPerFileReadBenchmark: files=$numFiles
rowsPerFile=$rowsPerFile rounds=$rounds " +
+ s"warmupRounds=$warmupRounds hadoopConfEntries=${hadoopConf.size()}")
+ results.foreach { case (name, micros) =>
+ println(f"ParquetPerFileReadBenchmark: path=$name cpuMicrosPerFile
median=${micros(micros.size / 2)}%.1f " +
+ f"min=${micros.head}%.1f max=${micros.last}%.1f")
+ }
+ // scalastyle:on println
+ } finally {
+ spark.stop()
+ }
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
index a7135b0352b7..620541a1deb6 100644
---
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
@@ -259,7 +259,7 @@ abstract class
Spark3LegacyHoodieParquetFileFormat(shouldAppendPartitionValues:
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
footerFileMetaData.getSchema)
}
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] =
if (shouldUseInternalSchema) {
- // Same guard as ParquetSchemaEvolutionUtils.getHadoopConfClone:
schema-on-read cannot
+ // Same guard as ParquetSchemaEvolutionUtils.getFileReadConf:
schema-on-read cannot
// reconstruct shredded variants, so fail loudly instead of silently
dropping typed_value.
// Empty projections (count(*)) read no column data and keep working.
if (requiredSchema.nonEmpty) {
diff --git
a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
index e5f94df47866..ad6f79bb74e1 100644
---
a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
+++
b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
@@ -164,13 +164,7 @@ class Spark33ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP,
0), 0)
val hadoopAttemptContext =
- new
TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData,
enableVectorizedReader), attemptId)
-
- // Try to push down filters when filter push-down is enabled.
- // Notice: This push-down is RowGroups level, not individual records.
- if (pushed.isDefined) {
-
ParquetInputFormat.setFilterPredicate(hadoopAttemptContext.getConfiguration,
pushed.get)
- }
+ new
TaskAttemptContextImpl(schemaEvolutionUtils.getFileReadConf(footerFileMetaData,
enableVectorizedReader, pushed), attemptId)
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
val vectorizedReader = schemaEvolutionUtils.buildVectorizedReader(
diff --git
a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
index f350afacfecf..a14956b40417 100644
---
a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
+++
b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
@@ -161,13 +161,7 @@ class Spark34ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP,
0), 0)
val hadoopAttemptContext =
- new
TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData,
enableVectorizedReader), attemptId)
-
- // Try to push down filters when filter push-down is enabled.
- // Notice: This push-down is RowGroups level, not individual records.
- if (pushed.isDefined) {
-
ParquetInputFormat.setFilterPredicate(hadoopAttemptContext.getConfiguration,
pushed.get)
- }
+ new
TaskAttemptContextImpl(schemaEvolutionUtils.getFileReadConf(footerFileMetaData,
enableVectorizedReader, pushed), attemptId)
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
val vectorizedReader = schemaEvolutionUtils.buildVectorizedReader(
diff --git
a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
index 31955d13dbdf..eacf3be55b7a 100644
---
a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
+++
b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
@@ -168,13 +168,7 @@ class Spark35ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP,
0), 0)
val hadoopAttemptContext =
- new
TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData,
enableVectorizedReader), attemptId)
-
- // Try to push down filters when filter push-down is enabled.
- // Notice: This push-down is RowGroups level, not individual records.
- if (pushed.isDefined) {
-
ParquetInputFormat.setFilterPredicate(hadoopAttemptContext.getConfiguration,
pushed.get)
- }
+ new
TaskAttemptContextImpl(schemaEvolutionUtils.getFileReadConf(footerFileMetaData,
enableVectorizedReader, pushed), attemptId)
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
val vectorizedReader = schemaEvolutionUtils.buildVectorizedReader(
diff --git
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40LegacyHoodieParquetFileFormat.scala
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40LegacyHoodieParquetFileFormat.scala
index 37980b6dfa41..36fd0b7b614f 100644
---
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40LegacyHoodieParquetFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40LegacyHoodieParquetFileFormat.scala
@@ -242,7 +242,7 @@ class Spark40LegacyHoodieParquetFileFormat(private val
shouldAppendPartitionValu
// Clone new conf
val hadoopAttemptConf = new
Configuration(broadcastedHadoopConf.value.value)
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] =
if (shouldUseInternalSchema) {
- // Same guard as ParquetSchemaEvolutionUtils.getHadoopConfClone:
schema-on-read cannot
+ // Same guard as ParquetSchemaEvolutionUtils.getFileReadConf:
schema-on-read cannot
// reconstruct shredded variants, so fail loudly instead of silently
dropping typed_value.
// Empty projections (count(*)) read no column data and keep working.
if (requiredSchema.nonEmpty) {
diff --git
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40ParquetReader.scala
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40ParquetReader.scala
index c98567a70936..d1ccc8139fb3 100644
---
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40ParquetReader.scala
+++
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40ParquetReader.scala
@@ -171,13 +171,7 @@ class Spark40ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP,
0), 0)
val hadoopAttemptContext =
- new
TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData,
enableVectorizedReader), attemptId)
-
- // Try to push down filters when filter push-down is enabled.
- // Notice: This push-down is RowGroups level, not individual records.
- if (pushed.isDefined) {
-
ParquetInputFormat.setFilterPredicate(hadoopAttemptContext.getConfiguration,
pushed.get)
- }
+ new
TaskAttemptContextImpl(schemaEvolutionUtils.getFileReadConf(footerFileMetaData,
enableVectorizedReader, pushed), attemptId)
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
val vectorizedReader = schemaEvolutionUtils.buildVectorizedReader(
diff --git
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41LegacyHoodieParquetFileFormat.scala
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41LegacyHoodieParquetFileFormat.scala
index 08a7e7816702..186d9bec76c3 100644
---
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41LegacyHoodieParquetFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41LegacyHoodieParquetFileFormat.scala
@@ -245,7 +245,7 @@ class Spark41LegacyHoodieParquetFileFormat(private val
shouldAppendPartitionValu
// Clone new conf
val hadoopAttemptConf = new
Configuration(broadcastedHadoopConf.value.value)
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]]
= if (shouldUseInternalSchema) {
- // Same guard as ParquetSchemaEvolutionUtils.getHadoopConfClone:
schema-on-read cannot
+ // Same guard as ParquetSchemaEvolutionUtils.getFileReadConf:
schema-on-read cannot
// reconstruct shredded variants, so fail loudly instead of silently
dropping typed_value.
// Empty projections (count(*)) read no column data and keep working.
if (requiredSchema.nonEmpty) {
diff --git
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41ParquetReader.scala
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41ParquetReader.scala
index 940fbb281e8f..f5b04301adfe 100644
---
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41ParquetReader.scala
+++
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41ParquetReader.scala
@@ -171,13 +171,7 @@ class Spark41ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP,
0), 0)
val hadoopAttemptContext =
- new
TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData,
enableVectorizedReader), attemptId)
-
- // Try to push down filters when filter push-down is enabled.
- // Notice: This push-down is RowGroups level, not individual records.
- pushed.foreach {
-
ParquetInputFormat.setFilterPredicate(hadoopAttemptContext.getConfiguration, _)
- }
+ new
TaskAttemptContextImpl(schemaEvolutionUtils.getFileReadConf(footerFileMetaData,
enableVectorizedReader, pushed), attemptId)
if (enableVectorizedReader) {
buildVectorizedIterator(
hadoopAttemptContext, split, file.partitionValues, partitionSchema,
convertTz,
diff --git
a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42LegacyHoodieParquetFileFormat.scala
b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42LegacyHoodieParquetFileFormat.scala
index 27c8b1f7b731..fe462522f6ac 100644
---
a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42LegacyHoodieParquetFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42LegacyHoodieParquetFileFormat.scala
@@ -245,7 +245,7 @@ class Spark42LegacyHoodieParquetFileFormat(private val
shouldAppendPartitionValu
// Clone new conf
val hadoopAttemptConf = new
Configuration(broadcastedHadoopConf.value.value)
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]]
= if (shouldUseInternalSchema) {
- // Same guard as ParquetSchemaEvolutionUtils.getHadoopConfClone:
schema-on-read cannot
+ // Same guard as ParquetSchemaEvolutionUtils.getFileReadConf:
schema-on-read cannot
// reconstruct shredded variants, so fail loudly instead of silently
dropping typed_value.
// Empty projections (count(*)) read no column data and keep working.
if (requiredSchema.nonEmpty) {
diff --git
a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42ParquetReader.scala
b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42ParquetReader.scala
index 9b265af6b98f..a3519a7a0337 100644
---
a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42ParquetReader.scala
+++
b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42ParquetReader.scala
@@ -171,13 +171,7 @@ class Spark42ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP,
0), 0)
val hadoopAttemptContext =
- new
TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData,
enableVectorizedReader), attemptId)
-
- // Try to push down filters when filter push-down is enabled.
- // Notice: This push-down is RowGroups level, not individual records.
- pushed.foreach {
-
ParquetInputFormat.setFilterPredicate(hadoopAttemptContext.getConfiguration, _)
- }
+ new
TaskAttemptContextImpl(schemaEvolutionUtils.getFileReadConf(footerFileMetaData,
enableVectorizedReader, pushed), attemptId)
if (enableVectorizedReader) {
buildVectorizedIterator(
hadoopAttemptContext, split, file.partitionValues, partitionSchema,
convertTz,
