Skip to content

[HUDI-9012] Implement and utilize native HFile writer #12866

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yihua merged 40 commits into from
May 25, 2025
Merged

[HUDI-9012] Implement and utilize native HFile writer #12866

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
40 commits
Select commit Hold shift + click to select a range
4399145
Import the initial write classes for HFile.
linliu-code Feb 21, 2025
c09362e
Enable HFileWriter to Hudi
linliu-code Feb 24, 2025
cef75d5
Address CI failures
linliu-code Feb 26, 2025
bdc3132
Add metablock
linliu-code Feb 27, 2025
22cf3ee
Fix bloom filter code
linliu-code Feb 27, 2025
24992df
Support bootstrap index
linliu-code Feb 28, 2025
0991053
Address comments
linliu-code Mar 4, 2025
fc16d80
Address some comments
linliu-code Mar 4, 2025
e94388f
Address comments
linliu-code Mar 7, 2025
cee4d3e
refacotr
linliu-code Mar 11, 2025
2e29038
Refactor more.
linliu-code Mar 11, 2025
af13aa8
Add test for same key location
linliu-code Mar 12, 2025
86942ab
Performance improvement:
linliu-code Mar 25, 2025
e3b126b
Add support:
linliu-code Mar 26, 2025
796a4e9
Fix a bug introduced.
linliu-code Mar 26, 2025
4d7f164
Address some comments
linliu-code Mar 26, 2025
4ff67f2
rebase
linliu-code May 16, 2025
6d21d65
Address more comments
linliu-code May 23, 2025
2e36d66
Address comments
linliu-code May 24, 2025
b0eaed4
Fix nits
yihua May 24, 2025
6f5973f
Move code for readability
yihua May 24, 2025
33853da
Fix nit
yihua May 24, 2025
b8e52bb
Fix getUncompressedBlockDataToWrite
yihua May 24, 2025
29fcdcd
Fix nits
yihua May 24, 2025
ac319c8
Fir more nits
yihua May 24, 2025
21a4ef0
Fix HFileMetaBlock
yihua May 25, 2025
63e48db
Fix test comments
linliu-code May 25, 2025
d498d7b
Fix nit
yihua May 25, 2025
90b0213
Fix build
yihua May 25, 2025
4b3d694
Address comments
yihua May 25, 2025
2ca40f2
Fix comparator class name
yihua May 25, 2025
e042eae
Address more comments
yihua May 25, 2025
1e4de72
Fix tests
linliu-code May 25, 2025
0d65ffc
Address more comments
yihua May 25, 2025
400908b
Fix nits
yihua May 25, 2025
75b9e85
Improve readability
yihua May 25, 2025
94a2105
Fix bug during refactoring
yihua May 25, 2025
4b6ba7b
Revert reader attribute class changes to make reader logic the same
yihua May 25, 2025
d0af9c5
Fix nit
yihua May 25, 2025
37de2d6
Fix tests
linliu-code May 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 2 additions & 5 deletions ...ommon/src/main/java/org/apache/hudi/common/bootstrap/index/hfile/HFileBootstrapIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.storage.HoodieStorage;
Expand Down Expand Up @@ -144,9 +143,7 @@ public BootstrapIndex.IndexReader createReader() {

@Override
public BootstrapIndex.IndexWriter createWriter(String bootstrapBasePath) {
return (IndexWriter) ReflectionUtils.loadClass("org.apache.hudi.common.bootstrap.index.hfile.HBaseHFileBootstrapIndexWriter",
new Class<?>[] {String.class, HoodieTableMetaClient.class},
bootstrapBasePath, metaClient);
return new HFileBootstrapIndexWriter(bootstrapBasePath, metaClient);
}

@Override
Expand All @@ -155,7 +152,7 @@ public void dropIndex() {
StoragePath[] indexPaths = new StoragePath[] {partitionIndexPath(metaClient), fileIdIndexPath(metaClient)};
for (StoragePath indexPath : indexPaths) {
if (metaClient.getStorage().exists(indexPath)) {
LOG.info("Dropping bootstrap index. Deleting file : " + indexPath);
LOG.info("Dropping bootstrap index. Deleting file: {}", indexPath);
metaClient.getStorage().deleteDirectory(indexPath);
}
}
Expand Down
59 changes: 24 additions & 35 deletions ...hfile/HBaseHFileBootstrapIndexWriter.java → ...ndex/hfile/HFileBootstrapIndexWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,51 +30,45 @@
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.io.hfile.HFileContext;
import org.apache.hudi.io.hfile.HFileWriter;
import org.apache.hudi.io.hfile.HFileWriterImpl;
import org.apache.hudi.storage.StoragePath;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.io.hfile.HFileContext;
import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static org.apache.hudi.common.bootstrap.index.hfile.HFileBootstrapIndex.INDEX_INFO_KEY;
import static org.apache.hudi.common.bootstrap.index.hfile.HFileBootstrapIndex.INDEX_INFO_KEY_STRING;
import static org.apache.hudi.common.bootstrap.index.hfile.HFileBootstrapIndex.fileIdIndexPath;
import static org.apache.hudi.common.bootstrap.index.hfile.HFileBootstrapIndex.getFileGroupKey;
import static org.apache.hudi.common.bootstrap.index.hfile.HFileBootstrapIndex.getPartitionKey;
import static org.apache.hudi.common.bootstrap.index.hfile.HFileBootstrapIndex.partitionIndexPath;
import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;

public class HBaseHFileBootstrapIndexWriter extends BootstrapIndex.IndexWriter {
private static final Logger LOG = LoggerFactory.getLogger(HBaseHFileBootstrapIndexWriter.class);
public class HFileBootstrapIndexWriter extends BootstrapIndex.IndexWriter {
private static final Logger LOG = LoggerFactory.getLogger(HFileBootstrapIndexWriter.class);

private final String bootstrapBasePath;
private final StoragePath indexByPartitionPath;
private final StoragePath indexByFileIdPath;
private HFile.Writer indexByPartitionWriter;
private HFile.Writer indexByFileIdWriter;
private HFileWriter indexByPartitionWriter;
private HFileWriter indexByFileIdWriter;

private boolean closed = false;
private int numPartitionKeysAdded = 0;
private int numFileIdKeysAdded = 0;

private final Map<String, List<BootstrapFileMapping>> sourceFileMappings = new HashMap<>();

public HBaseHFileBootstrapIndexWriter(String bootstrapBasePath, HoodieTableMetaClient metaClient) {
public HFileBootstrapIndexWriter(String bootstrapBasePath, HoodieTableMetaClient metaClient) {
super(metaClient);
try {
metaClient.initializeBootstrapDirsIfNotExists();
Expand Down Expand Up @@ -114,9 +108,7 @@ private void writeNextPartition(String partitionPath, String bootstrapPartitionP
m.getBootstrapFileStatus())).collect(Collectors.toMap(Pair::getKey, Pair::getValue)));
Option<byte[]> bytes = TimelineMetadataUtils.serializeAvroMetadata(bootstrapPartitionMetadata, HoodieBootstrapPartitionMetadata.class);
if (bytes.isPresent()) {
indexByPartitionWriter
.append(new KeyValue(getUTF8Bytes(getPartitionKey(partitionPath)), new byte[0], new byte[0],
HConstants.LATEST_TIMESTAMP, KeyValue.Type.Put, bytes.get()));
indexByPartitionWriter.append(getPartitionKey(partitionPath), bytes.get());
Copy link
Contributor

@yihua yihua May 21, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When writing the key-value pairs, do you write the bytes confirming to HConstants.LATEST_TIMESTAMP, KeyValue.Type.Put, etc.?

Copy link
Collaborator Author

@linliu-code linliu-code May 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since these fields are not used in Hudi, I did not check.

Copy link
Collaborator Author

@linliu-code linliu-code May 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LATEST_TIMESTAMP : 0x7fffffffffffffffL, KeyValue.Type.Put: 4. These fields are not contained in the keys. For backward compatability, we could support them.

Copy link
Collaborator Author

@linliu-code linliu-code May 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Want to confirm: Is it ok for us to add such support in a followup pr?

Copy link
Contributor

@yihua yihua May 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As long as the native HFile reader and writer work seamlessly together without any correctness issue, we can land this PR. However, we need to address the compatibility with HBase as a blocker for Hudi 1.1 release so that the HBase-based HFile reader in Hudi 0.x can read the HFile written by the Hudi 1.1 release (for backwards compatibility).

Copy link
Collaborator Author

@linliu-code linliu-code May 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To address that we need to run some tests to use hbase reader to read files generated by native writer. I will do that after this pr.

Copy link
Collaborator Author

@linliu-code linliu-code May 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually let me file a ticket to track this.https://issues.apache.org/jira/browse/HUDI-9449

numPartitionKeysAdded++;
}
} catch (IOException e) {
Expand All @@ -135,11 +127,10 @@ private void writeNextSourceFileMapping(BootstrapFileMapping mapping) {
srcFilePartitionInfo.setPartitionPath(mapping.getPartitionPath());
srcFilePartitionInfo.setBootstrapPartitionPath(mapping.getBootstrapPartitionPath());
srcFilePartitionInfo.setBootstrapFileStatus(mapping.getBootstrapFileStatus());
KeyValue kv = new KeyValue(getUTF8Bytes(getFileGroupKey(mapping.getFileGroupId())), new byte[0], new byte[0],
HConstants.LATEST_TIMESTAMP, KeyValue.Type.Put,
TimelineMetadataUtils.serializeAvroMetadata(srcFilePartitionInfo,
HoodieBootstrapFilePartitionInfo.class).get());
indexByFileIdWriter.append(kv);
indexByFileIdWriter.append(
getFileGroupKey(mapping.getFileGroupId()),
TimelineMetadataUtils.serializeAvroMetadata(
srcFilePartitionInfo, HoodieBootstrapFilePartitionInfo.class).get());
numFileIdKeysAdded++;
} catch (IOException e) {
throw new HoodieIOException(e.getMessage(), e);
Expand All @@ -166,9 +157,11 @@ private void commit() {
.build();
LOG.info("Appending FileId FileInfo :" + fileIdIndexInfo);

indexByPartitionWriter.appendFileInfo(INDEX_INFO_KEY,
indexByPartitionWriter.appendFileInfo(
INDEX_INFO_KEY_STRING,
TimelineMetadataUtils.serializeAvroMetadata(partitionIndexInfo, HoodieBootstrapIndexInfo.class).get());
indexByFileIdWriter.appendFileInfo(INDEX_INFO_KEY,
indexByFileIdWriter.appendFileInfo(
INDEX_INFO_KEY_STRING,
TimelineMetadataUtils.serializeAvroMetadata(fileIdIndexInfo, HoodieBootstrapIndexInfo.class).get());

close();
Expand Down Expand Up @@ -196,15 +189,11 @@ public void close() {
@Override
public void begin() {
try {
HFileContext meta = new HFileContextBuilder().withCellComparator(new org.apache.hudi.common.bootstrap.index.HFileBootstrapIndex.HoodieKVComparator()).build();
this.indexByPartitionWriter = HFile.getWriterFactory(metaClient.getStorageConf().unwrapAs(Configuration.class),
new CacheConfig(metaClient.getStorageConf().unwrapAs(Configuration.class)))
.withPath((FileSystem) metaClient.getStorage().getFileSystem(), new Path(indexByPartitionPath.toUri()))
.withFileContext(meta).create();
this.indexByFileIdWriter = HFile.getWriterFactory(metaClient.getStorageConf().unwrapAs(Configuration.class),
new CacheConfig(metaClient.getStorageConf().unwrapAs(Configuration.class)))
.withPath((FileSystem) metaClient.getStorage().getFileSystem(), new Path(indexByFileIdPath.toUri()))
.withFileContext(meta).create();
HFileContext context = HFileContext.builder().build();
Copy link
Contributor

@yihua yihua May 21, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you verify the KV comparator class name written to the HFile by the new HFile writer? We need this to be backwards compatible, i.e., Hudi 1.1 release writing table version 6 and 8, both of which should be read without any problem using Hudi 0.14/15 and Hudi 1.0.2 release.

Copy link
Collaborator Author

@linliu-code linliu-code May 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Will verify.

Copy link
Collaborator Author

@linliu-code linliu-code May 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After remove hbase dependency, all comparator will be gone. What class name should we give?

Copy link
Collaborator Author

@linliu-code linliu-code May 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds like org.apache.hadoop.hbase.util.Bytes.ByteArrayComparator should work since it compares two byte arrays without interpreting the underlying bytes and this is what we do.

Copy link
Collaborator Author

@linliu-code linliu-code May 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed the class name to org.apache.hadoop.hbase.util.Bytes.ByteArrayComparator now; following test will demonstrate if it works for hbase reader.

Copy link
Contributor

@yihua yihua May 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the CellComparator implementation class. We should use org.apache.hudi.io.storage.HoodieHBaseKVComparator.

OutputStream outputStreamForPartitionWriter = metaClient.getStorage().create(indexByPartitionPath);
this.indexByPartitionWriter = new HFileWriterImpl(context, outputStreamForPartitionWriter);
OutputStream outputStreamForFileIdWriter = metaClient.getStorage().create(indexByFileIdPath);
this.indexByFileIdWriter = new HFileWriterImpl(context, outputStreamForFileIdWriter);
} catch (IOException ioe) {
throw new HoodieIOException(ioe.getMessage(), ioe);
}
Expand Down
73 changes: 29 additions & 44 deletions ...g/apache/hudi/common/util/HFileUtils.java → ...g/apache/hudi/common/util/HFileUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,29 +29,25 @@
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.io.compress.CompressionCodec;
import org.apache.hudi.io.hfile.HFileContext;
import org.apache.hudi.io.hfile.HFileWriter;
import org.apache.hudi.io.hfile.HFileWriterImpl;
import org.apache.hudi.io.storage.HoodieAvroHFileReaderImplBase;
import org.apache.hudi.io.storage.HoodieFileReader;
import org.apache.hudi.io.storage.HoodieHBaseKVComparator;
import org.apache.hudi.io.storage.HoodieIOFactory;
import org.apache.hudi.keygen.BaseKeyGenerator;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.io.hfile.HFileContext;
import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
Expand All @@ -70,17 +66,15 @@ public class HFileUtils extends FileFormatUtils {
private static final int DEFAULT_BLOCK_SIZE_FOR_LOG_FILE = 1024 * 1024;

/**
* Gets the {@link Compression.Algorithm} Enum based on the {@link CompressionCodec} name.
*
* @param paramsMap parameter map containing the compression codec config.
* @return the {@link Compression.Algorithm} Enum.
* @return the {@link CompressionCodec} Enum.
*/
public static Compression.Algorithm getHFileCompressionAlgorithm(Map<String, String> paramsMap) {
String algoName = paramsMap.get(HFILE_COMPRESSION_ALGORITHM_NAME.key());
if (StringUtils.isNullOrEmpty(algoName)) {
return Compression.Algorithm.GZ;
public static CompressionCodec getHFileCompressionAlgorithm(Map<String, String> paramsMap) {
String codecName = paramsMap.get(HFILE_COMPRESSION_ALGORITHM_NAME.key());
if (StringUtils.isNullOrEmpty(codecName)) {
return CompressionCodec.GZIP;
}
return Compression.Algorithm.valueOf(algoName.toUpperCase());
return CompressionCodec.findCodecByName(codecName);
}

@Override
Expand Down Expand Up @@ -167,17 +161,9 @@ public ByteArrayOutputStream serializeRecordsToLogBlock(HoodieStorage storage,
Schema readerSchema,
String keyFieldName,
Map<String, String> paramsMap) throws IOException {
Compression.Algorithm compressionAlgorithm = getHFileCompressionAlgorithm(paramsMap);
HFileContext context = new HFileContextBuilder()
.withBlockSize(DEFAULT_BLOCK_SIZE_FOR_LOG_FILE)
.withCompression(compressionAlgorithm)
.withCellComparator(new HoodieHBaseKVComparator())
.build();

Configuration conf = storage.getConf().unwrapAs(Configuration.class);
CacheConfig cacheConfig = new CacheConfig(conf);
CompressionCodec compressionCodec = getHFileCompressionAlgorithm(paramsMap);
ByteArrayOutputStream baos = new ByteArrayOutputStream();
FSDataOutputStream ostream = new FSDataOutputStream(baos, null);
OutputStream ostream = new DataOutputStream(baos);

// Use simple incrementing counter as a key
boolean useIntegerKey = !getRecordKey(records.get(0), readerSchema, keyFieldName).isPresent();
Expand Down Expand Up @@ -209,26 +195,25 @@ public ByteArrayOutputStream serializeRecordsToLogBlock(HoodieStorage storage,
sortedRecordsMap.put(recordKey, recordBytes);
}

HFile.Writer writer = HFile.getWriterFactory(conf, cacheConfig)
.withOutputStream(ostream).withFileContext(context).create();

// Write the records
sortedRecordsMap.forEach((recordKey, recordBytes) -> {
try {
KeyValue kv = new KeyValue(recordKey.getBytes(), null, null, recordBytes);
writer.append(kv);
} catch (IOException e) {
throw new HoodieIOException("IOException serializing records", e);
}
});

writer.appendFileInfo(
getUTF8Bytes(HoodieAvroHFileReaderImplBase.SCHEMA_KEY), getUTF8Bytes(readerSchema.toString()));
HFileContext context = HFileContext.builder()
.blockSize(DEFAULT_BLOCK_SIZE_FOR_LOG_FILE)
.compressionCodec(compressionCodec)
.build();
try (HFileWriter writer = new HFileWriterImpl(context, ostream)) {
sortedRecordsMap.forEach((recordKey,recordBytes) -> {
try {
writer.append(recordKey, recordBytes);
} catch (IOException e) {
throw new HoodieIOException("IOException serializing records", e);
}
});
writer.appendFileInfo(
HoodieAvroHFileReaderImplBase.SCHEMA_KEY,
getUTF8Bytes(readerSchema.toString()));
}

writer.close();
ostream.flush();
ostream.close();

return baos;
}

Expand Down
11 changes: 6 additions & 5 deletions ...ache/hudi/common/util/TestHFileUtils.java → ...ache/hudi/common/util/TestHFileUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,8 @@

package org.apache.hudi.common.util;

import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hudi.io.compress.CompressionCodec;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
Expand All @@ -36,8 +37,8 @@
*/
public class TestHFileUtils {
@ParameterizedTest
@EnumSource(Compression.Algorithm.class)
public void testGetHFileCompressionAlgorithm(Compression.Algorithm algo) {
@EnumSource(CompressionCodec.class)
public void testGetHFileCompressionAlgorithm(CompressionCodec algo) {
for (boolean upperCase : new boolean[] {true, false}) {
Map<String, String> paramsMap = Collections.singletonMap(
HFILE_COMPRESSION_ALGORITHM_NAME.key(),
Expand All @@ -48,12 +49,12 @@ public void testGetHFileCompressionAlgorithm(Compression.Algorithm algo) {

@Test
public void testGetHFileCompressionAlgorithmWithEmptyString() {
assertEquals(Compression.Algorithm.GZ, getHFileCompressionAlgorithm(
assertEquals(CompressionCodec.GZIP, getHFileCompressionAlgorithm(
Collections.singletonMap(HFILE_COMPRESSION_ALGORITHM_NAME.key(), "")));
}

@Test
public void testGetDefaultHFileCompressionAlgorithm() {
assertEquals(Compression.Algorithm.GZ, getHFileCompressionAlgorithm(Collections.emptyMap()));
assertEquals(CompressionCodec.GZIP, getHFileCompressionAlgorithm(Collections.emptyMap()));
}
}
15 changes: 5 additions & 10 deletions hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieAvroFileWriterFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.io.compress.CompressionCodec;
import org.apache.hudi.io.storage.HoodieAvroHFileReaderImplBase;
import org.apache.hudi.io.storage.HoodieFileWriter;
import org.apache.hudi.io.storage.HoodieFileWriterFactory;
Expand All @@ -38,7 +39,6 @@
import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.orc.CompressionKind;
import org.apache.parquet.avro.AvroSchemaConverter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
Expand All @@ -48,11 +48,6 @@
import java.io.OutputStream;
import java.util.Properties;

import static org.apache.hudi.io.hadoop.HoodieHFileConfig.CACHE_DATA_IN_L1;
import static org.apache.hudi.io.hadoop.HoodieHFileConfig.DROP_BEHIND_CACHE_COMPACTION;
import static org.apache.hudi.io.hadoop.HoodieHFileConfig.HFILE_COMPARATOR;
import static org.apache.hudi.io.hadoop.HoodieHFileConfig.PREFETCH_ON_OPEN;

public class HoodieAvroFileWriterFactory extends HoodieFileWriterFactory {

public HoodieAvroFileWriterFactory(HoodieStorage storage) {
Expand Down Expand Up @@ -98,13 +93,13 @@ protected HoodieFileWriter newHFileFileWriter(
String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
BloomFilter filter = createBloomFilter(config);
HoodieHFileConfig hfileConfig = new HoodieHFileConfig(storage.getConf().unwrapAs(Configuration.class),
Compression.Algorithm.valueOf(
HoodieHFileConfig hfileConfig = new HoodieHFileConfig(
storage.getConf(),
CompressionCodec.findCodecByName(
config.getString(HoodieStorageConfig.HFILE_COMPRESSION_ALGORITHM_NAME)),
config.getInt(HoodieStorageConfig.HFILE_BLOCK_SIZE),
config.getLong(HoodieStorageConfig.HFILE_MAX_FILE_SIZE),
HoodieAvroHFileReaderImplBase.KEY_FIELD_NAME,
PREFETCH_ON_OPEN, CACHE_DATA_IN_L1, DROP_BEHIND_CACHE_COMPACTION, filter, HFILE_COMPARATOR);
HoodieAvroHFileReaderImplBase.KEY_FIELD_NAME, filter);
return new HoodieAvroHFileWriter(instantTime, path, hfileConfig, schema, taskContextSupplier, config.getBoolean(HoodieTableConfig.POPULATE_META_FIELDS));
}

Expand Down
Loading
Loading