You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2025/09/22 05:01:24 UTC
(hudi) branch master updated: [HUDI-9777] Avoid redundant file length fetch in HoodieNativeHFileReader (#13797)
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 66d2780d8983 [HUDI-9777] Avoid redundant file length fetch in HoodieNativeHFileReader (#13797)
66d2780d8983 is described below
commit 66d2780d8983d89d5981339ae2307c49febaca3a
Author: vamsikarnika <va...@gmail.com>
AuthorDate: Mon Sep 22 10:31:17 2025 +0530
[HUDI-9777] Avoid redundant file length fetch in HoodieNativeHFileReader (#13797)
Co-authored-by: Vamsi <va...@onehouse.ai>
Co-authored-by: Y Ethan Guo <et...@gmail.com>
---
.../apache/hudi/avro/HoodieAvroReaderContext.java | 63 ++++++++++++++++++----
.../apache/hudi/io/storage/HFileReaderFactory.java | 27 +++++++---
.../hudi/io/storage/HoodieFileReaderFactory.java | 20 +++++++
.../io/hadoop/HoodieAvroFileReaderFactory.java | 10 ++++
4 files changed, 104 insertions(+), 16 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroReaderContext.java b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroReaderContext.java
index c90dc0186741..c659c6beecaf 100644
--- a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroReaderContext.java
+++ b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroReaderContext.java
@@ -39,12 +39,14 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.SizeEstimator;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.expression.Predicate;
import org.apache.hudi.io.storage.HoodieAvroFileReader;
import org.apache.hudi.io.storage.HoodieIOFactory;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.StoragePathInfo;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
@@ -55,6 +57,7 @@ import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.function.Function;
import static org.apache.hudi.common.config.HoodieReaderConfig.RECORD_MERGE_IMPL_CLASSES_WRITE_CONFIG_KEY;
import static org.apache.hudi.common.util.ValidationUtils.checkState;
@@ -127,6 +130,24 @@ public class HoodieAvroReaderContext extends HoodieReaderContext<IndexedRecord>
this.isMultiFormat = tableConfig.isMultipleBaseFileFormatsEnabled();
}
+ @Override
+ public ClosableIterator<IndexedRecord> getFileRecordIterator(
+ StoragePathInfo storagePathInfo, long start, long length, Schema dataSchema, Schema requiredSchema,
+ HoodieStorage storage) throws IOException {
+ boolean isLogFile = FSUtils.isLogFile(storagePathInfo.getPath());
+ HoodieAvroFileReader reader = getOrCreateFileReader(storagePathInfo.getPath(), isLogFile, format -> {
+ try {
+ return (HoodieAvroFileReader) HoodieIOFactory.getIOFactory(storage)
+ .getReaderFactory(HoodieRecord.HoodieRecordType.AVRO).getFileReader(new HoodieConfig(),
+ storagePathInfo, format, Option.empty());
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to create avro records iterator from file path " + storagePathInfo.getPath(), e);
+ }
+ });
+
+ return getFileRecordIterator(reader, storagePathInfo.getPath(), isLogFile, dataSchema, requiredSchema);
+ }
+
@Override
public ClosableIterator<IndexedRecord> getFileRecordIterator(
StoragePath filePath,
@@ -135,8 +156,40 @@ public class HoodieAvroReaderContext extends HoodieReaderContext<IndexedRecord>
Schema dataSchema,
Schema requiredSchema,
HoodieStorage storage) throws IOException {
- HoodieAvroFileReader reader;
boolean isLogFile = FSUtils.isLogFile(filePath);
+ HoodieAvroFileReader reader = getOrCreateFileReader(filePath, isLogFile, format -> {
+ try {
+ return (HoodieAvroFileReader) HoodieIOFactory.getIOFactory(storage)
+ .getReaderFactory(HoodieRecord.HoodieRecordType.AVRO).getFileReader(new HoodieConfig(),
+ filePath, format, Option.empty());
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to create avro records iterator from file path " + filePath, e);
+ }
+ });
+
+ return getFileRecordIterator(reader, filePath, isLogFile, dataSchema, requiredSchema);
+ }
+
+ private HoodieAvroFileReader getOrCreateFileReader(
+ StoragePath path, boolean isLogFile, Function<HoodieFileFormat, HoodieAvroFileReader> func) throws IOException {
+ if (reusableFileReaders.containsKey(path)) {
+ return reusableFileReaders.get(path);
+ } else {
+ HoodieFileFormat fileFormat = isMultiFormat && !isLogFile ? HoodieFileFormat.fromFileExtension(path.getFileExtension()) : baseFileFormat;
+ try {
+ return func.apply(fileFormat);
+ } catch (HoodieIOException e) {
+ throw e.getIOException();
+ }
+ }
+ }
+
+ public ClosableIterator<IndexedRecord> getFileRecordIterator(
+ HoodieAvroFileReader reader,
+ StoragePath filePath,
+ boolean isLogFile,
+ Schema dataSchema,
+ Schema requiredSchema) throws IOException {
Schema fileOutputSchema;
Map<String, String> renamedColumns;
if (isLogFile) {
@@ -147,14 +200,6 @@ public class HoodieAvroReaderContext extends HoodieReaderContext<IndexedRecord>
fileOutputSchema = requiredSchemaForFileAndRenamedColumns.getLeft();
renamedColumns = requiredSchemaForFileAndRenamedColumns.getRight();
}
- if (reusableFileReaders.containsKey(filePath)) {
- reader = reusableFileReaders.get(filePath);
- } else {
- HoodieFileFormat fileFormat = isMultiFormat && !isLogFile ? HoodieFileFormat.fromFileExtension(filePath.getFileExtension()) : baseFileFormat;
- reader = (HoodieAvroFileReader) HoodieIOFactory.getIOFactory(storage)
- .getReaderFactory(HoodieRecord.HoodieRecordType.AVRO).getFileReader(new HoodieConfig(),
- filePath, fileFormat, Option.empty());
- }
if (keyFilterOpt.isEmpty()) {
return reader.getIndexedRecordIterator(dataSchema, fileOutputSchema, renamedColumns);
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HFileReaderFactory.java b/hudi-common/src/main/java/org/apache/hudi/io/storage/HFileReaderFactory.java
index 1eb84469739a..c6a8c83d29bb 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HFileReaderFactory.java
+++ b/hudi-common/src/main/java/org/apache/hudi/io/storage/HFileReaderFactory.java
@@ -48,20 +48,26 @@ public class HFileReaderFactory {
private final HoodieMetadataConfig metadataConfig;
private final TypedProperties properties;
private final Either<StoragePath, byte[]> fileSource;
+ private Option<Long> fileSizeOpt;
- public HFileReaderFactory(HoodieStorage storage,
- TypedProperties properties,
- Either<StoragePath, byte[]> fileSource) {
+ private HFileReaderFactory(HoodieStorage storage,
+ TypedProperties properties,
+ Either<StoragePath, byte[]> fileSource,
+ Option<Long> fileSizeOpt) {
this.storage = storage;
this.metadataConfig = HoodieMetadataConfig.newBuilder().withProperties(properties).build();
this.properties = properties;
this.fileSource = fileSource;
+ this.fileSizeOpt = fileSizeOpt;
}
public HFileReader createHFileReader() throws IOException {
- final long fileSize = determineFileSize();
+ if (fileSizeOpt.isEmpty()) {
+ fileSizeOpt = Option.of(determineFileSize());
+ }
+ final long fileSize = fileSizeOpt.get();
final SeekableDataInputStream inputStream = createInputStream(fileSize);
-
+
if (shouldEnableBlockCaching()) {
int blockCacheSize = ConfigUtils.getIntWithAltKeys(
properties, HoodieReaderConfig.HFILE_BLOCK_CACHE_SIZE);
@@ -70,7 +76,7 @@ public class HFileReaderFactory {
String filePath = getFilePath();
return new CachingHFileReaderImpl(inputStream, fileSize, filePath, blockCacheSize, cacheTtlMinutes);
}
-
+
return new HFileReaderImpl(inputStream, fileSize);
}
@@ -119,6 +125,7 @@ public class HFileReaderFactory {
private HoodieStorage storage;
private Option<TypedProperties> properties = Option.empty();
private Either<StoragePath, byte[]> fileSource;
+ private Option<Long> fileSizeOpt = Option.empty();
public Builder withStorage(HoodieStorage storage) {
this.storage = storage;
@@ -142,11 +149,17 @@ public class HFileReaderFactory {
return this;
}
+ public Builder withFileSize(long fileSize) {
+ ValidationUtils.checkState(fileSize >= 0, "file size is invalid, should be greater than or equal to zero");
+ this.fileSizeOpt = Option.of(fileSize);
+ return this;
+ }
+
public HFileReaderFactory build() {
ValidationUtils.checkArgument(storage != null, "Storage cannot be null");
ValidationUtils.checkArgument(fileSource != null, "HFile source cannot be null");
TypedProperties props = properties.isPresent() ? properties.get() : new TypedProperties();
- return new HFileReaderFactory(storage, props, fileSource);
+ return new HFileReaderFactory(storage, props, fileSource, fileSizeOpt);
}
}
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileReaderFactory.java b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileReaderFactory.java
index 0c63a3057c21..912fc6a76a0c 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileReaderFactory.java
+++ b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileReaderFactory.java
@@ -25,6 +25,7 @@ import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.StoragePathInfo;
import org.apache.avro.Schema;
@@ -77,6 +78,20 @@ public class HoodieFileReaderFactory {
}
}
+ public HoodieFileReader getFileReader(HoodieConfig hoodieConfig, StoragePathInfo pathInfo, HoodieFileFormat format,
+ Option<Schema> schemaOption) throws IOException {
+ switch (format) {
+ case PARQUET:
+ return newParquetFileReader(pathInfo.getPath());
+ case HFILE:
+ return newHFileFileReader(hoodieConfig, pathInfo, schemaOption);
+ case ORC:
+ return newOrcFileReader(pathInfo.getPath());
+ default:
+ throw new UnsupportedOperationException(format + " format not supported yet.");
+ }
+ }
+
public HoodieFileReader getContentReader(HoodieConfig hoodieConfig, StoragePath path, HoodieFileFormat format,
HoodieStorage storage, byte[] content,
Option<Schema> schemaOption) throws IOException {
@@ -97,6 +112,11 @@ public class HoodieFileReaderFactory {
throw new UnsupportedOperationException();
}
+ protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig, StoragePathInfo pathInfo,
+ Option<Schema> schemaOption) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig, StoragePath path,
HoodieStorage storage, byte[] content, Option<Schema> schemaOption)
throws IOException {
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieAvroFileReaderFactory.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieAvroFileReaderFactory.java
index c209ac0f8600..11db6de9de82 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieAvroFileReaderFactory.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieAvroFileReaderFactory.java
@@ -28,6 +28,7 @@ import org.apache.hudi.io.storage.HoodieFileReaderFactory;
import org.apache.hudi.io.storage.HoodieNativeAvroHFileReader;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.StoragePathInfo;
import org.apache.avro.Schema;
@@ -55,6 +56,15 @@ public class HoodieAvroFileReaderFactory extends HoodieFileReaderFactory {
.readerFactory(readerFactory).path(path).schema(schemaOption).build();
}
+ protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig, StoragePathInfo pathInfo,
+ Option<Schema> schemaOption) {
+ HFileReaderFactory readerFactory = HFileReaderFactory.builder()
+ .withStorage(storage).withProps(hoodieConfig.getProps())
+ .withPath(pathInfo.getPath()).withFileSize(pathInfo.getLength()).build();
+ return HoodieNativeAvroHFileReader.builder()
+ .readerFactory(readerFactory).path(pathInfo.getPath()).schema(schemaOption).build();
+ }
+
@Override
protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
StoragePath path,