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 2024/05/15 15:25:49 UTC
(hudi) 03/13: [HUDI-7589] Add API to create HoodieStorage in HoodieIOFactory (#11208)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit e8f9c1f1cef5bb0ea6c6ed728f88419c93bd370e
Author: Jon Vexler <jb...@gmail.com>
AuthorDate: Wed May 15 06:56:57 2024 -0700
[HUDI-7589] Add API to create HoodieStorage in HoodieIOFactory (#11208)
Co-authored-by: Jonathan Vexler <=>
---
.../main/java/org/apache/hudi/cli/HoodieCLI.java | 4 +-
.../hudi/cli/commands/TestRepairsCommand.java | 6 +--
.../org/apache/hudi/index/HoodieIndexUtils.java | 2 +-
.../java/org/apache/hudi/io/HoodieMergeHandle.java | 4 +-
.../java/org/apache/hudi/io/HoodieReadHandle.java | 8 +--
.../table/action/commit/HoodieMergeHelper.java | 8 +--
.../hudi/client/utils/TestCommitMetadataUtils.java | 4 +-
.../GenericRecordValidationTestUtils.java | 2 +-
.../run/strategy/JavaExecutionStrategy.java | 4 +-
.../hudi/client/TestJavaHoodieBackedMetadata.java | 8 +--
.../MultipleSparkJobExecutionStrategy.java | 10 ++--
.../strategy/SingleSparkJobExecutionStrategy.java | 4 +-
.../io/storage/HoodieSparkFileReaderFactory.java | 21 ++++----
.../io/storage/HoodieSparkFileWriterFactory.java | 20 ++++---
.../hudi/io/storage/HoodieSparkIOFactory.java | 14 +++--
.../bootstrap/ParquetBootstrapMetadataHandler.java | 4 +-
.../java/org/apache/hudi/client/TestMultiFS.java | 6 +--
.../functional/TestHoodieBackedMetadata.java | 8 +--
.../functional/TestHoodieBackedTableMetadata.java | 5 +-
.../hudi/testutils/FunctionalTestHarness.java | 3 +-
.../hudi/testutils/HoodieClientTestUtils.java | 5 +-
.../hudi/common/table/HoodieTableMetaClient.java | 7 ++-
.../table/log/block/HoodieHFileDataBlock.java | 6 +--
.../table/log/block/HoodieParquetDataBlock.java | 2 +-
.../hudi/io/storage/HoodieFileReaderFactory.java | 46 ++++++++--------
.../hudi/io/storage/HoodieFileWriterFactory.java | 29 +++++-----
.../apache/hudi/io/storage/HoodieIOFactory.java | 24 +++++++--
.../hudi/metadata/HoodieBackedTableMetadata.java | 2 +-
.../hudi/metadata/HoodieTableMetadataUtil.java | 9 ++--
.../apache/hudi/storage/HoodieStorageUtils.java | 49 ++---------------
.../hudi/common/testutils/HoodieTestUtils.java | 13 +----
.../hudi/sink/clustering/ClusteringOperator.java | 4 +-
.../sink/TestStreamWriteOperatorCoordinator.java | 9 ++--
.../hudi/table/catalog/HoodieCatalogTestUtils.java | 7 +++
.../hudi/table/catalog/TestHoodieCatalog.java | 4 +-
.../hudi/table/catalog/TestHoodieHiveCatalog.java | 15 +++---
.../apache/hudi/table/format/TestInputFormat.java | 7 +--
.../test/java/org/apache/hudi/utils/TestUtils.java | 17 +++---
.../org/apache/hudi/common/util/HFileUtils.java | 1 -
.../org/apache/hudi/hadoop/fs/HadoopFSUtils.java | 3 +-
.../io/hadoop/HoodieAvroFileReaderFactory.java | 46 ++++++----------
.../io/hadoop/HoodieAvroFileWriterFactory.java | 61 +++++++---------------
.../hudi/io/storage/HoodieHadoopIOFactory.java | 40 ++++++++++++--
.../hudi/storage/hadoop/HoodieHadoopStorage.java | 26 ++++-----
.../fs/TestFSUtilsWithRetryWrapperEnable.java | 6 +--
.../common/fs/TestHoodieWrapperFileSystem.java | 4 +-
.../common/functional/TestHoodieLogFormat.java | 5 +-
.../TestHoodieLogFormatAppendFailure.java | 4 +-
.../table/timeline/TestHoodieActiveTimeline.java | 4 +-
.../io/hadoop/TestHoodieAvroFileReaderFactory.java | 6 +--
.../hudi/io/hadoop/TestHoodieOrcReaderWriter.java | 2 +-
.../hudi/hadoop/HoodieHFileRecordReader.java | 2 +-
.../apache/hudi/hadoop/SchemaEvolutionContext.java | 4 +-
.../utils/HoodieRealtimeRecordReaderUtils.java | 2 +-
.../hive/TestHoodieCombineHiveInputFormat.java | 4 +-
.../TestHoodieMergeOnReadSnapshotReader.java | 7 +--
.../TestHoodieMergeOnReadTableInputFormat.java | 4 +-
.../realtime/TestHoodieRealtimeRecordReader.java | 4 +-
.../hudi/hadoop/testutils/InputFormatTestUtil.java | 4 +-
.../reader/DFSHoodieDatasetInputReader.java | 1 -
.../org/apache/hudi/storage/HoodieStorage.java | 6 +++
.../scala/org/apache/hudi/HoodieBaseRelation.scala | 4 +-
.../src/test/java/HoodieJavaStreamingApp.java | 3 +-
.../hudi/functional/TestMORDataSourceStorage.scala | 4 +-
.../hudi/procedure/TestClusteringProcedure.scala | 8 +--
.../hudi/procedure/TestCompactionProcedure.scala | 2 +-
.../org/apache/hudi/hive/TestHiveSyncTool.java | 3 +-
.../apache/hudi/hive/testutils/HiveTestUtil.java | 4 +-
.../utilities/HoodieMetadataTableValidator.java | 5 +-
.../hudi/utilities/deltastreamer/DeltaSync.java | 4 +-
.../deltastreamer/HoodieDeltaStreamer.java | 6 +--
.../apache/hudi/utilities/streamer/StreamSync.java | 4 +-
.../TestHoodieDeltaStreamerWithMultiWriter.java | 7 +--
.../streamer/TestStreamSyncUnitTests.java | 6 +--
.../utilities/testutils/UtilitiesTestBase.java | 4 +-
75 files changed, 348 insertions(+), 362 deletions(-)
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/HoodieCLI.java b/hudi-cli/src/main/java/org/apache/hudi/cli/HoodieCLI.java
index a71aa8fc05e..0f99701d1ae 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/HoodieCLI.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/HoodieCLI.java
@@ -26,8 +26,8 @@ import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -86,7 +86,7 @@ public class HoodieCLI {
if (storage == null || force) {
storage = (tableMetadata != null)
? tableMetadata.getStorage()
- : HoodieStorageUtils.getStorage(FileSystem.get(conf.unwrap()));
+ : new HoodieHadoopStorage(FileSystem.get(conf.unwrap()));
}
}
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java
index 5b62bf1b2cf..681cc2be0d1 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java
@@ -40,8 +40,8 @@ import org.apache.hudi.common.util.PartitionPathEncodeUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.keygen.SimpleKeyGenerator;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.testutils.Assertions;
import org.apache.avro.generic.GenericRecord;
@@ -141,7 +141,7 @@ public class TestRepairsCommand extends CLIFunctionalTestHarness {
assertTrue(ShellEvaluationResultUtil.isSuccess(result));
// expected all 'No'.
- String[][] rows = FSUtils.getAllPartitionFoldersThreeLevelsDown(HoodieStorageUtils.getStorage(fs), tablePath)
+ String[][] rows = FSUtils.getAllPartitionFoldersThreeLevelsDown(new HoodieHadoopStorage(fs), tablePath)
.stream()
.map(partition -> new String[] {partition, "No", "None"})
.toArray(String[][]::new);
@@ -171,7 +171,7 @@ public class TestRepairsCommand extends CLIFunctionalTestHarness {
Object result = shell.evaluate(() -> "repair addpartitionmeta --dryrun false");
assertTrue(ShellEvaluationResultUtil.isSuccess(result));
- List<String> paths = FSUtils.getAllPartitionFoldersThreeLevelsDown(HoodieStorageUtils.getStorage(fs), tablePath);
+ List<String> paths = FSUtils.getAllPartitionFoldersThreeLevelsDown(new HoodieHadoopStorage(fs), tablePath);
// after dry run, the action will be 'Repaired'
String[][] rows = paths.stream()
.map(partition -> new String[] {partition, "No", "Repaired"})
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
index db32112750a..e4d0269a3e6 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
@@ -188,7 +188,7 @@ public class HoodieIndexUtils {
List<String> foundRecordKeys = new ArrayList<>();
try (HoodieFileReader fileReader = HoodieIOFactory.getIOFactory(configuration)
.getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, configuration, filePath)) {
+ .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, filePath)) {
// Load all rowKeys from the file, to double-confirm
if (!candidateRecordKeys.isEmpty()) {
HoodieTimer timer = HoodieTimer.start();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
index 3c3a820ab09..56c183c34e2 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
@@ -462,9 +462,9 @@ public class HoodieMergeHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O>
}
long oldNumWrites = 0;
- try (HoodieFileReader reader = HoodieIOFactory.getIOFactory(storage.getConf())
+ try (HoodieFileReader reader = HoodieIOFactory.getIOFactory(hoodieTable.getStorageConf())
.getReaderFactory(this.recordMerger.getRecordType())
- .getFileReader(config, hoodieTable.getStorageConf(), oldFilePath)) {
+ .getFileReader(config, oldFilePath)) {
oldNumWrites = reader.getTotalRecords();
} catch (IOException e) {
throw new HoodieUpsertException("Failed to check for merge data validation", e);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieReadHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieReadHandle.java
index 01678b68e96..71d691ad580 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieReadHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieReadHandle.java
@@ -69,12 +69,12 @@ public abstract class HoodieReadHandle<T, I, K, O> extends HoodieIOHandle<T, I,
}
protected HoodieFileReader createNewFileReader() throws IOException {
- return HoodieIOFactory.getIOFactory(storage.getConf()).getReaderFactory(this.config.getRecordMerger().getRecordType())
- .getFileReader(config, hoodieTable.getStorageConf(), getLatestBaseFile().getStoragePath());
+ return HoodieIOFactory.getIOFactory(hoodieTable.getStorageConf()).getReaderFactory(this.config.getRecordMerger().getRecordType())
+ .getFileReader(config, getLatestBaseFile().getStoragePath());
}
protected HoodieFileReader createNewFileReader(HoodieBaseFile hoodieBaseFile) throws IOException {
- return HoodieIOFactory.getIOFactory(storage.getConf()).getReaderFactory(this.config.getRecordMerger().getRecordType())
- .getFileReader(config, hoodieTable.getStorageConf(), hoodieBaseFile.getStoragePath());
+ return HoodieIOFactory.getIOFactory(hoodieTable.getStorageConf()).getReaderFactory(this.config.getRecordMerger().getRecordType())
+ .getFileReader(config, hoodieBaseFile.getStoragePath());
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
index a13253bc1b0..cfd9ff606dd 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
@@ -82,7 +82,7 @@ public class HoodieMergeHelper<T> extends BaseMergeHelper {
HoodieRecord.HoodieRecordType recordType = table.getConfig().getRecordMerger().getRecordType();
HoodieFileReader baseFileReader = HoodieIOFactory.getIOFactory(storageConf)
.getReaderFactory(recordType)
- .getFileReader(writeConfig, storageConf, mergeHandle.getOldFilePath());
+ .getFileReader(writeConfig, mergeHandle.getOldFilePath());
HoodieFileReader bootstrapFileReader = null;
Schema writerSchema = mergeHandle.getWriterSchemaWithMetaFields();
@@ -112,10 +112,10 @@ public class HoodieMergeHelper<T> extends BaseMergeHelper {
if (baseFile.getBootstrapBaseFile().isPresent()) {
StoragePath bootstrapFilePath = baseFile.getBootstrapBaseFile().get().getStoragePath();
StorageConfiguration<?> bootstrapFileConfig = table.getStorageConf().newInstance();
- bootstrapFileReader = HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(recordType).newBootstrapFileReader(
+ bootstrapFileReader = HoodieIOFactory.getIOFactory(bootstrapFileConfig).getReaderFactory(recordType).newBootstrapFileReader(
baseFileReader,
- HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(recordType)
- .getFileReader(writeConfig, bootstrapFileConfig, bootstrapFilePath),
+ HoodieIOFactory.getIOFactory(bootstrapFileConfig).getReaderFactory(recordType)
+ .getFileReader(writeConfig, bootstrapFilePath),
mergeHandle.getPartitionFields(),
mergeHandle.getPartitionValues());
recordSchema = mergeHandle.getWriterSchemaWithMetaFields();
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/utils/TestCommitMetadataUtils.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/utils/TestCommitMetadataUtils.java
index 9fa7780b6b6..d8cd9d22050 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/utils/TestCommitMetadataUtils.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/utils/TestCommitMetadataUtils.java
@@ -34,8 +34,8 @@ import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.marker.WriteMarkers;
@@ -92,7 +92,7 @@ public class TestCommitMetadataUtils extends HoodieCommonTestHarness {
when(table.getMetaClient()).thenReturn(metaClient);
Mockito.when(table.getConfig()).thenReturn(writeConfig);
when(metaClient.getTableType()).thenReturn(HoodieTableType.MERGE_ON_READ);
- when(metaClient.getStorage()).thenReturn(HoodieStorageUtils.getStorage(fileSystem));
+ when(metaClient.getStorage()).thenReturn(new HoodieHadoopStorage(fileSystem));
when(metaClient.getBasePath()).thenReturn(basePath);
when(metaClient.getMarkerFolderPath(any())).thenReturn(basePath + ".hoodie/.temp");
when(table.getContext()).thenReturn(context);
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/GenericRecordValidationTestUtils.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/GenericRecordValidationTestUtils.java
index 34972f01832..1b1bb6bcfaa 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/GenericRecordValidationTestUtils.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/GenericRecordValidationTestUtils.java
@@ -148,7 +148,7 @@ public class GenericRecordValidationTestUtils {
StorageConfiguration storageConf = HadoopFSUtils.getStorageConf(conf);
try (HoodieAvroHFileReaderImplBase reader = (HoodieAvroHFileReaderImplBase)
HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(HoodieRecord.HoodieRecordType.AVRO)
- .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, storageConf, new StoragePath(path), HoodieFileFormat.HFILE)) {
+ .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, new StoragePath(path), HoodieFileFormat.HFILE)) {
valuesAsList.addAll(HoodieAvroHFileReaderImplBase.readAllRecords(reader)
.stream().map(e -> (GenericRecord) e).collect(Collectors.toList()));
} catch (IOException e) {
diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/JavaExecutionStrategy.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/JavaExecutionStrategy.java
index 5b216807932..6dd0fc09d72 100644
--- a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/JavaExecutionStrategy.java
+++ b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/JavaExecutionStrategy.java
@@ -193,7 +193,7 @@ public abstract class JavaExecutionStrategy<T>
baseFileReader = StringUtils.isNullOrEmpty(clusteringOp.getDataFilePath())
? Option.empty()
: Option.of(HoodieIOFactory.getIOFactory(table.getStorageConf()).getReaderFactory(recordType)
- .getFileReader(config, table.getStorageConf(), new StoragePath(clusteringOp.getDataFilePath())));
+ .getFileReader(config, new StoragePath(clusteringOp.getDataFilePath())));
HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
Iterator<HoodieRecord<T>> fileSliceReader = new HoodieFileSliceReader(baseFileReader, scanner, readerSchema, tableConfig.getPreCombineField(), writeConfig.getRecordMerger(),
tableConfig.getProps(),
@@ -222,7 +222,7 @@ public abstract class JavaExecutionStrategy<T>
List<HoodieRecord<T>> records = new ArrayList<>();
clusteringOps.forEach(clusteringOp -> {
try (HoodieFileReader baseFileReader = HoodieIOFactory.getIOFactory(getHoodieTable().getStorageConf()).getReaderFactory(recordType)
- .getFileReader(getHoodieTable().getConfig(), getHoodieTable().getStorageConf(), new StoragePath(clusteringOp.getDataFilePath()))) {
+ .getFileReader(getHoodieTable().getConfig(), new StoragePath(clusteringOp.getDataFilePath()))) {
Schema readerSchema = HoodieAvroUtils.addMetadataFields(new Schema.Parser().parse(getWriteConfig().getSchema()));
Iterator<HoodieRecord> recordIterator = baseFileReader.getRecordIterator(readerSchema);
// NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
diff --git a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java
index c2413133477..1c26fb82001 100644
--- a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java
+++ b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java
@@ -544,8 +544,8 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase {
List<FileSlice> fileSlices = table.getSliceView().getLatestFileSlices("files").collect(Collectors.toList());
HoodieBaseFile baseFile = fileSlices.get(0).getBaseFile().get();
HoodieAvroHFileReaderImplBase hoodieHFileReader = (HoodieAvroHFileReaderImplBase)
- HoodieIOFactory.getIOFactory(context.getStorageConf()).getReaderFactory(HoodieRecordType.AVRO).getFileReader(
- writeConfig, context.getStorageConf(), new StoragePath(baseFile.getPath()));
+ HoodieIOFactory.getIOFactory(context.getStorageConf()).getReaderFactory(HoodieRecordType.AVRO)
+ .getFileReader(writeConfig, new StoragePath(baseFile.getPath()));
List<IndexedRecord> records = HoodieAvroHFileReaderImplBase.readAllRecords(hoodieHFileReader);
records.forEach(entry -> {
if (populateMetaFields) {
@@ -971,8 +971,8 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase {
final HoodieBaseFile baseFile = fileSlices.get(0).getBaseFile().get();
HoodieAvroHFileReaderImplBase hoodieHFileReader = (HoodieAvroHFileReaderImplBase)
- HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(HoodieRecordType.AVRO).getFileReader(
- table.getConfig(), context.getStorageConf(), new StoragePath(baseFile.getPath()));
+ HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(HoodieRecordType.AVRO)
+ .getFileReader(table.getConfig(), new StoragePath(baseFile.getPath()));
List<IndexedRecord> records = HoodieAvroHFileReaderImplBase.readAllRecords(hoodieHFileReader);
records.forEach(entry -> {
if (enableMetaFields) {
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
index fe1e6710673..3182b2f9a66 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
@@ -380,8 +380,8 @@ public abstract class MultipleSparkJobExecutionStrategy<T>
private HoodieFileReader getBaseOrBootstrapFileReader(StorageConfiguration<?> storageConf, String bootstrapBasePath, Option<String[]> partitionFields, ClusteringOperation clusteringOp)
throws IOException {
- HoodieFileReader baseFileReader = getHoodieSparkIOFactory().getReaderFactory(recordType)
- .getFileReader(writeConfig, storageConf, new StoragePath(clusteringOp.getDataFilePath()));
+ HoodieFileReader baseFileReader = getHoodieSparkIOFactory(storageConf).getReaderFactory(recordType)
+ .getFileReader(writeConfig, new StoragePath(clusteringOp.getDataFilePath()));
// handle bootstrap path
if (StringUtils.nonEmpty(clusteringOp.getBootstrapFilePath()) && StringUtils.nonEmpty(bootstrapBasePath)) {
String bootstrapFilePath = clusteringOp.getBootstrapFilePath();
@@ -392,10 +392,10 @@ public abstract class MultipleSparkJobExecutionStrategy<T>
partitionValues = getPartitionFieldVals(partitionFields, partitionFilePath, bootstrapBasePath, baseFileReader.getSchema(),
storageConf.unwrapAs(Configuration.class));
}
- baseFileReader = getHoodieSparkIOFactory().getReaderFactory(recordType).newBootstrapFileReader(
+ baseFileReader = getHoodieSparkIOFactory(storageConf).getReaderFactory(recordType).newBootstrapFileReader(
baseFileReader,
- getHoodieSparkIOFactory().getReaderFactory(recordType).getFileReader(
- writeConfig, storageConf, new StoragePath(bootstrapFilePath)), partitionFields,
+ getHoodieSparkIOFactory(storageConf).getReaderFactory(recordType).getFileReader(
+ writeConfig, new StoragePath(bootstrapFilePath)), partitionFields,
partitionValues);
}
return baseFileReader;
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/SingleSparkJobExecutionStrategy.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/SingleSparkJobExecutionStrategy.java
index 06ba64dad89..a7faca1a418 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/SingleSparkJobExecutionStrategy.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/SingleSparkJobExecutionStrategy.java
@@ -147,8 +147,8 @@ public abstract class SingleSparkJobExecutionStrategy<T>
Schema readerSchema = HoodieAvroUtils.addMetadataFields(new Schema.Parser().parse(getWriteConfig().getSchema()));
Iterable<HoodieRecord<T>> indexedRecords = () -> {
try {
- HoodieFileReader baseFileReader = getHoodieSparkIOFactory().getReaderFactory(recordType)
- .getFileReader(writeConfig, getHoodieTable().getStorageConf(), new StoragePath(clusteringOp.getDataFilePath()));
+ HoodieFileReader baseFileReader = getHoodieSparkIOFactory(getHoodieTable().getStorageConf()).getReaderFactory(recordType)
+ .getFileReader(writeConfig, new StoragePath(clusteringOp.getDataFilePath()));
Option<BaseKeyGenerator> keyGeneratorOp = HoodieSparkKeyGeneratorFactory.createBaseKeyGenerator(writeConfig);
// NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
// payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java
index b28718f3c73..06b33c8dded 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java
@@ -31,29 +31,32 @@ import java.io.IOException;
public class HoodieSparkFileReaderFactory extends HoodieFileReaderFactory {
+ public HoodieSparkFileReaderFactory(StorageConfiguration<?> storageConf) {
+ super(storageConf);
+ }
+
@Override
- public HoodieFileReader newParquetFileReader(StorageConfiguration<?> conf, StoragePath path) {
- conf.setIfUnset(SQLConf.PARQUET_BINARY_AS_STRING().key(), SQLConf.PARQUET_BINARY_AS_STRING().defaultValueString());
- conf.setIfUnset(SQLConf.PARQUET_INT96_AS_TIMESTAMP().key(), SQLConf.PARQUET_INT96_AS_TIMESTAMP().defaultValueString());
- conf.setIfUnset(SQLConf.CASE_SENSITIVE().key(), SQLConf.CASE_SENSITIVE().defaultValueString());
+ public HoodieFileReader newParquetFileReader(StoragePath path) {
+ storageConf.setIfUnset(SQLConf.PARQUET_BINARY_AS_STRING().key(), SQLConf.PARQUET_BINARY_AS_STRING().defaultValueString());
+ storageConf.setIfUnset(SQLConf.PARQUET_INT96_AS_TIMESTAMP().key(), SQLConf.PARQUET_INT96_AS_TIMESTAMP().defaultValueString());
+ storageConf.setIfUnset(SQLConf.CASE_SENSITIVE().key(), SQLConf.CASE_SENSITIVE().defaultValueString());
// Using string value of this conf to preserve compatibility across spark versions.
- conf.setIfUnset("spark.sql.legacy.parquet.nanosAsLong", "false");
+ storageConf.setIfUnset("spark.sql.legacy.parquet.nanosAsLong", "false");
// This is a required config since Spark 3.4.0: SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED
// Using string value of this conf to preserve compatibility across spark versions.
- conf.setIfUnset("spark.sql.parquet.inferTimestampNTZ.enabled", "true");
- return new HoodieSparkParquetReader(conf, path);
+ storageConf.setIfUnset("spark.sql.parquet.inferTimestampNTZ.enabled", "true");
+ return new HoodieSparkParquetReader(storageConf, path);
}
@Override
protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
- StorageConfiguration<?> conf,
StoragePath path,
Option<Schema> schemaOption) throws IOException {
throw new HoodieIOException("Not support read HFile");
}
@Override
- protected HoodieFileReader newOrcFileReader(StorageConfiguration<?> conf, StoragePath path) {
+ protected HoodieFileReader newOrcFileReader(StoragePath path) {
throw new HoodieIOException("Not support read orc file");
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java
index ff17b48bf0c..6a513e2d7d6 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java
@@ -42,9 +42,13 @@ import java.io.OutputStream;
public class HoodieSparkFileWriterFactory extends HoodieFileWriterFactory {
+ public HoodieSparkFileWriterFactory(StorageConfiguration<?> storageConf) {
+ super(storageConf);
+ }
+
@Override
protected HoodieFileWriter newParquetFileWriter(
- String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
+ String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
boolean populateMetaFields = config.getBooleanOrDefault(HoodieTableConfig.POPULATE_META_FIELDS);
String compressionCodecName = config.getStringOrDefault(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME);
@@ -52,14 +56,14 @@ public class HoodieSparkFileWriterFactory extends HoodieFileWriterFactory {
if (compressionCodecName.isEmpty()) {
compressionCodecName = null;
}
- HoodieRowParquetWriteSupport writeSupport = getHoodieRowParquetWriteSupport(conf, schema,
+ HoodieRowParquetWriteSupport writeSupport = getHoodieRowParquetWriteSupport(storageConf, schema,
config, enableBloomFilter(populateMetaFields, config));
HoodieRowParquetConfig parquetConfig = new HoodieRowParquetConfig(writeSupport,
CompressionCodecName.fromConf(compressionCodecName),
config.getIntOrDefault(HoodieStorageConfig.PARQUET_BLOCK_SIZE),
config.getIntOrDefault(HoodieStorageConfig.PARQUET_PAGE_SIZE),
config.getLongOrDefault(HoodieStorageConfig.PARQUET_MAX_FILE_SIZE),
- conf.unwrapAs(Configuration.class),
+ storageConf.unwrapAs(Configuration.class),
config.getDoubleOrDefault(HoodieStorageConfig.PARQUET_COMPRESSION_RATIO_FRACTION),
config.getBooleanOrDefault(HoodieStorageConfig.PARQUET_DICTIONARY_ENABLED));
parquetConfig.getHadoopConf().addResource(writeSupport.getHadoopConf());
@@ -67,10 +71,10 @@ public class HoodieSparkFileWriterFactory extends HoodieFileWriterFactory {
return new HoodieSparkParquetWriter(path, parquetConfig, instantTime, taskContextSupplier, populateMetaFields);
}
- protected HoodieFileWriter newParquetFileWriter(
- OutputStream outputStream, StorageConfiguration<?> conf, HoodieConfig config, Schema schema) throws IOException {
+ protected HoodieFileWriter newParquetFileWriter(OutputStream outputStream, HoodieConfig config,
+ Schema schema) throws IOException {
boolean enableBloomFilter = false;
- HoodieRowParquetWriteSupport writeSupport = getHoodieRowParquetWriteSupport(conf, schema, config, enableBloomFilter);
+ HoodieRowParquetWriteSupport writeSupport = getHoodieRowParquetWriteSupport(storageConf, schema, config, enableBloomFilter);
String compressionCodecName = config.getStringOrDefault(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME);
// Support PARQUET_COMPRESSION_CODEC_NAME is ""
if (compressionCodecName.isEmpty()) {
@@ -88,13 +92,13 @@ public class HoodieSparkFileWriterFactory extends HoodieFileWriterFactory {
}
@Override
- protected HoodieFileWriter newHFileFileWriter(String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
+ protected HoodieFileWriter newHFileFileWriter(String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
throw new HoodieIOException("Not support write to HFile");
}
@Override
- protected HoodieFileWriter newOrcFileWriter(String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
+ protected HoodieFileWriter newOrcFileWriter(String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
throw new HoodieIOException("Not support write to Orc file");
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkIOFactory.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkIOFactory.java
index 16431d61551..9d673b98908 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkIOFactory.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkIOFactory.java
@@ -20,21 +20,25 @@
package org.apache.hudi.io.storage;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.storage.StorageConfiguration;
/**
* Creates readers and writers for SPARK and AVRO record payloads
*/
public class HoodieSparkIOFactory extends HoodieHadoopIOFactory {
- private static final HoodieSparkIOFactory HOODIE_SPARK_IO_FACTORY = new HoodieSparkIOFactory();
- public static HoodieSparkIOFactory getHoodieSparkIOFactory() {
- return HOODIE_SPARK_IO_FACTORY;
+ public HoodieSparkIOFactory(StorageConfiguration<?> storageConf) {
+ super(storageConf);
+ }
+
+ public static HoodieSparkIOFactory getHoodieSparkIOFactory(StorageConfiguration<?> storageConf) {
+ return new HoodieSparkIOFactory(storageConf);
}
@Override
public HoodieFileReaderFactory getReaderFactory(HoodieRecord.HoodieRecordType recordType) {
if (recordType == HoodieRecord.HoodieRecordType.SPARK) {
- return new HoodieSparkFileReaderFactory();
+ return new HoodieSparkFileReaderFactory(storageConf);
}
return super.getReaderFactory(recordType);
}
@@ -42,7 +46,7 @@ public class HoodieSparkIOFactory extends HoodieHadoopIOFactory {
@Override
public HoodieFileWriterFactory getWriterFactory(HoodieRecord.HoodieRecordType recordType) {
if (recordType == HoodieRecord.HoodieRecordType.SPARK) {
- return new HoodieSparkFileWriterFactory();
+ return new HoodieSparkFileWriterFactory(storageConf);
}
return super.getWriterFactory(recordType);
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/ParquetBootstrapMetadataHandler.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/ParquetBootstrapMetadataHandler.java
index adc6a456ac9..565551505c6 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/ParquetBootstrapMetadataHandler.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/ParquetBootstrapMetadataHandler.java
@@ -82,8 +82,8 @@ class ParquetBootstrapMetadataHandler extends BaseBootstrapMetadataHandler {
Schema schema) throws Exception {
HoodieRecord.HoodieRecordType recordType = table.getConfig().getRecordMerger().getRecordType();
- HoodieFileReader reader = getHoodieSparkIOFactory().getReaderFactory(recordType)
- .getFileReader(table.getConfig(), table.getStorageConf(), sourceFilePath);
+ HoodieFileReader reader = getHoodieSparkIOFactory(table.getStorageConf()).getReaderFactory(recordType)
+ .getFileReader(table.getConfig(), sourceFilePath);
HoodieExecutor<Void> executor = null;
try {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestMultiFS.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestMultiFS.java
index 2d29e936d15..007097a0a6c 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestMultiFS.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestMultiFS.java
@@ -32,7 +32,7 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
-import org.apache.hudi.storage.HoodieStorageUtils;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.testutils.HoodieClientTestUtils;
import org.apache.hudi.testutils.HoodieSparkClientTestHarness;
@@ -134,7 +134,7 @@ public class TestMultiFS extends HoodieSparkClientTestHarness {
// Read from hdfs
FileSystem fs = HadoopFSUtils.getFs(dfsBasePath, HoodieTestUtils.getDefaultStorageConf());
- HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(HoodieStorageUtils.getStorageConf(fs.getConf()), dfsBasePath);
+ HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(HadoopFSUtils.getStorageConf(fs.getConf()), dfsBasePath);
HoodieTimeline timeline = new HoodieActiveTimeline(metaClient).getCommitTimeline();
Dataset<Row> readRecords = HoodieClientTestUtils.readCommit(dfsBasePath, sqlContext, timeline, readCommitTime);
assertEquals(readRecords.count(), records.size());
@@ -155,7 +155,7 @@ public class TestMultiFS extends HoodieSparkClientTestHarness {
LOG.info("Reading from path: " + tablePath);
fs = HadoopFSUtils.getFs(tablePath, HoodieTestUtils.getDefaultStorageConf());
- metaClient = HoodieTestUtils.createMetaClient(fs.getConf(), tablePath);
+ metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(fs.getConf()), tablePath);
timeline = new HoodieActiveTimeline(metaClient).getCommitTimeline();
Dataset<Row> localReadRecords =
HoodieClientTestUtils.readCommit(tablePath, sqlContext, timeline, writeCommitTime);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
index a83fcd4bf27..30b1b63998d 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
@@ -821,8 +821,8 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
List<FileSlice> fileSlices = table.getSliceView().getLatestFileSlices("files").collect(Collectors.toList());
HoodieBaseFile baseFile = fileSlices.get(0).getBaseFile().get();
HoodieAvroHFileReaderImplBase hoodieHFileReader = (HoodieAvroHFileReaderImplBase)
- getHoodieSparkIOFactory().getReaderFactory(HoodieRecordType.AVRO).getFileReader(
- table.getConfig(), context.getStorageConf(), new StoragePath(baseFile.getPath()));
+ getHoodieSparkIOFactory(context.getStorageConf()).getReaderFactory(HoodieRecordType.AVRO).getFileReader(
+ table.getConfig(), new StoragePath(baseFile.getPath()));
List<IndexedRecord> records = HoodieAvroHFileReaderImplBase.readAllRecords(hoodieHFileReader);
records.forEach(entry -> {
if (populateMetaFields) {
@@ -1354,9 +1354,9 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
}
final HoodieBaseFile baseFile = fileSlices.get(0).getBaseFile().get();
- HoodieAvroHFileReaderImplBase hoodieHFileReader = (HoodieAvroHFileReaderImplBase) getHoodieSparkIOFactory()
+ HoodieAvroHFileReaderImplBase hoodieHFileReader = (HoodieAvroHFileReaderImplBase) getHoodieSparkIOFactory(context.getStorageConf())
.getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(table.getConfig(), context.getStorageConf(), new StoragePath(baseFile.getPath()));
+ .getFileReader(table.getConfig(), new StoragePath(baseFile.getPath()));
List<IndexedRecord> records = HoodieAvroHFileReaderImplBase.readAllRecords(hoodieHFileReader);
records.forEach(entry -> {
if (enableMetaFields) {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java
index c4a79f1ea71..e5824b02b03 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java
@@ -527,8 +527,9 @@ public class TestHoodieBackedTableMetadata extends TestHoodieMetadataBase {
final HoodieBaseFile baseFile = fileSlices.get(0).getBaseFile().get();
HoodieAvroHFileReaderImplBase hoodieHFileReader = (HoodieAvroHFileReaderImplBase)
- getHoodieSparkIOFactory().getReaderFactory(HoodieRecordType.AVRO).getFileReader(
- table.getConfig(), context.getStorageConf(), new StoragePath(baseFile.getPath()));
+ getHoodieSparkIOFactory(context.getStorageConf())
+ .getReaderFactory(HoodieRecordType.AVRO)
+ .getFileReader(table.getConfig(), new StoragePath(baseFile.getPath()));
List<IndexedRecord> records = HoodieAvroHFileReaderImplBase.readAllRecords(hoodieHFileReader);
records.forEach(entry -> {
assertNull(((GenericRecord) entry).get(HoodieRecord.RECORD_KEY_METADATA_FIELD));
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/FunctionalTestHarness.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/FunctionalTestHarness.java
index fa604e8edf5..488b7e170d5 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/FunctionalTestHarness.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/FunctionalTestHarness.java
@@ -33,6 +33,7 @@ import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.testutils.providers.DFSProvider;
import org.apache.hudi.testutils.providers.HoodieMetaClientProvider;
import org.apache.hudi.testutils.providers.HoodieWriteClientProvider;
@@ -151,7 +152,7 @@ public class FunctionalTestHarness implements SparkProvider, DFSProvider, Hoodie
hdfsTestService = new HdfsTestService();
dfsCluster = hdfsTestService.start(true);
- storage = HoodieStorageUtils.getStorage(dfsCluster.getFileSystem());
+ storage = new HoodieHadoopStorage(dfsCluster.getFileSystem());
storage.createDirectory(new StoragePath("/tmp"));
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java
index 90a33417277..3eb1da0eae9 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java
@@ -42,6 +42,7 @@ import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.timeline.service.TimelineService;
import org.apache.avro.generic.GenericRecord;
@@ -313,7 +314,7 @@ public class HoodieClientTestUtils {
* @return a new {@link HoodieTableMetaClient} instance.
*/
public static HoodieTableMetaClient createMetaClient(JavaSparkContext jsc, String basePath) {
- return HoodieTestUtils.createMetaClient(jsc.hadoopConfiguration(), basePath);
+ return HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(jsc.hadoopConfiguration()), basePath);
}
/**
@@ -322,7 +323,7 @@ public class HoodieClientTestUtils {
* @return a new {@link HoodieTableMetaClient} instance.
*/
public static HoodieTableMetaClient createMetaClient(SparkSession spark, String basePath) {
- return HoodieTestUtils.createMetaClient(spark.sessionState().newHadoopConf(), basePath);
+ return HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(spark.sessionState().newHadoopConf()), basePath);
}
private static Option<HoodieCommitMetadata> getCommitMetadataForInstant(HoodieTableMetaClient metaClient, HoodieInstant instant) {
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
index 4105677e03d..319cbdfbb4a 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
@@ -65,6 +65,7 @@ import java.util.stream.Stream;
import static org.apache.hudi.common.util.ConfigUtils.containsConfigProperty;
import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
+import static org.apache.hudi.io.storage.HoodieIOFactory.getIOFactory;
/**
* <code>HoodieTableMetaClient</code> allows to access meta-data about a hoodie table It returns meta-data about
@@ -302,9 +303,7 @@ public class HoodieTableMetaClient implements Serializable {
consistencyGuardConfig)
: new NoOpConsistencyGuard();
- storage = HoodieStorageUtils.getStorage(
- metaPath,
- getStorageConf(),
+ storage = getIOFactory(getStorageConf()).getStorage(metaPath,
fileSystemRetryConfig.isFileSystemActionRetryEnable(),
fileSystemRetryConfig.getMaxRetryIntervalMs(),
fileSystemRetryConfig.getMaxRetryNumbers(),
@@ -320,7 +319,7 @@ public class HoodieTableMetaClient implements Serializable {
}
public HoodieStorage getRawHoodieStorage() {
- return HoodieStorageUtils.getRawStorage(getStorage());
+ return getStorage().getRawStorage();
}
public StorageConfiguration<?> getStorageConf() {
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java
index f3b79e05787..356bab33bd0 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java
@@ -194,8 +194,8 @@ public class HoodieHFileDataBlock extends HoodieDataBlock {
// Read the content
try (HoodieFileReader reader = HoodieIOFactory.getIOFactory(storageConf)
.getReaderFactory(HoodieRecordType.AVRO)
- .getContentReader(hFileReaderConfig, storageConf, pathForReader,
- HoodieFileFormat.HFILE, storage, content, Option.of(getSchemaFromHeader()))) {
+ .getContentReader(hFileReaderConfig, pathForReader, HoodieFileFormat.HFILE,
+ storage, content, Option.of(getSchemaFromHeader()))) {
return unsafeCast(reader.getRecordIterator(readerSchema));
}
}
@@ -217,7 +217,7 @@ public class HoodieHFileDataBlock extends HoodieDataBlock {
try (final HoodieAvroHFileReaderImplBase reader = (HoodieAvroHFileReaderImplBase) HoodieIOFactory.getIOFactory(inlineConf)
.getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(hFileReaderConfig, inlineConf, inlinePath, HoodieFileFormat.HFILE, Option.of(getSchemaFromHeader()))) {
+ .getFileReader(hFileReaderConfig, inlinePath, HoodieFileFormat.HFILE, Option.of(getSchemaFromHeader()))) {
// Get writer's schema from the header
final ClosableIterator<HoodieRecord<IndexedRecord>> recordIterator =
fullKey ? reader.getRecordsByKeysIterator(sortedKeys, readerSchema) : reader.getRecordsByKeyPrefixIterator(sortedKeys, readerSchema);
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java
index 32f4f46a955..e370b156be8 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java
@@ -151,7 +151,7 @@ public class HoodieParquetDataBlock extends HoodieDataBlock {
Schema writerSchema = new Schema.Parser().parse(this.getLogBlockHeader().get(HeaderMetadataType.SCHEMA));
ClosableIterator<HoodieRecord<T>> iterator = HoodieIOFactory.getIOFactory(inlineConf).getReaderFactory(type)
- .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, inlineConf, inlineLogFilePath, PARQUET, Option.empty())
+ .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, inlineLogFilePath, PARQUET, Option.empty())
.getRecordIterator(writerSchema, readerSchema);
return iterator;
}
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 8637c468fdd..18dd976798d 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
@@ -40,71 +40,71 @@ import static org.apache.hudi.common.model.HoodieFileFormat.PARQUET;
*/
public class HoodieFileReaderFactory {
- public HoodieFileReader getFileReader(HoodieConfig hoodieConfig, StorageConfiguration<?> conf, StoragePath path) throws IOException {
+ protected final StorageConfiguration<?> storageConf;
+ public HoodieFileReaderFactory(StorageConfiguration<?> storageConf) {
+ this.storageConf = storageConf;
+ }
+
+ public HoodieFileReader getFileReader(HoodieConfig hoodieConfig, StoragePath path) throws IOException {
final String extension = FSUtils.getFileExtension(path.toString());
if (PARQUET.getFileExtension().equals(extension)) {
- return getFileReader(hoodieConfig, conf, path, PARQUET, Option.empty());
+ return getFileReader(hoodieConfig, path, PARQUET, Option.empty());
}
if (HFILE.getFileExtension().equals(extension)) {
- return getFileReader(hoodieConfig, conf, path, HFILE, Option.empty());
+ return getFileReader(hoodieConfig, path, HFILE, Option.empty());
}
if (ORC.getFileExtension().equals(extension)) {
- return getFileReader(hoodieConfig, conf, path, ORC, Option.empty());
+ return getFileReader(hoodieConfig, path, ORC, Option.empty());
}
throw new UnsupportedOperationException(extension + " format not supported yet.");
}
- public HoodieFileReader getFileReader(HoodieConfig hoodieConfig, StorageConfiguration<?> conf, StoragePath path, HoodieFileFormat format)
+ public HoodieFileReader getFileReader(HoodieConfig hoodieConfig, StoragePath path, HoodieFileFormat format)
throws IOException {
- return getFileReader(hoodieConfig, conf, path, format, Option.empty());
+ return getFileReader(hoodieConfig, path, format, Option.empty());
}
- public HoodieFileReader getFileReader(HoodieConfig hoodieConfig,
- StorageConfiguration<?> conf, StoragePath path, HoodieFileFormat format,
+ public HoodieFileReader getFileReader(HoodieConfig hoodieConfig, StoragePath path, HoodieFileFormat format,
Option<Schema> schemaOption) throws IOException {
switch (format) {
case PARQUET:
- return newParquetFileReader(conf, path);
+ return newParquetFileReader(path);
case HFILE:
- return newHFileFileReader(hoodieConfig, conf, path, schemaOption);
+ return newHFileFileReader(hoodieConfig, path, schemaOption);
case ORC:
- return newOrcFileReader(conf, path);
+ return newOrcFileReader(path);
default:
throw new UnsupportedOperationException(format + " format not supported yet.");
}
}
- public HoodieFileReader getContentReader(HoodieConfig hoodieConfig,
- StorageConfiguration<?> conf, StoragePath path, HoodieFileFormat format,
+ public HoodieFileReader getContentReader(HoodieConfig hoodieConfig, StoragePath path, HoodieFileFormat format,
HoodieStorage storage, byte[] content,
Option<Schema> schemaOption) throws IOException {
switch (format) {
case HFILE:
- return newHFileFileReader(hoodieConfig, conf, path, storage, content, schemaOption);
+ return newHFileFileReader(hoodieConfig, path, storage, content, schemaOption);
default:
throw new UnsupportedOperationException(format + " format not supported yet.");
}
}
- protected HoodieFileReader newParquetFileReader(StorageConfiguration<?> conf, StoragePath path) {
+ protected HoodieFileReader newParquetFileReader(StoragePath path) {
throw new UnsupportedOperationException();
}
- protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
- StorageConfiguration<?> conf, StoragePath path,
+ protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig, StoragePath path,
Option<Schema> schemaOption) throws IOException {
throw new UnsupportedOperationException();
}
- protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
- StorageConfiguration<?> conf, StoragePath path,
- HoodieStorage storage,
- byte[] content, Option<Schema> schemaOption)
+ protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig, StoragePath path,
+ HoodieStorage storage, byte[] content, Option<Schema> schemaOption)
throws IOException {
throw new UnsupportedOperationException();
}
- protected HoodieFileReader newOrcFileReader(StorageConfiguration<?> conf, StoragePath path) {
+ protected HoodieFileReader newOrcFileReader(StoragePath path) {
throw new UnsupportedOperationException();
}
@@ -118,4 +118,4 @@ public class HoodieFileReaderFactory {
protected static boolean isUseNativeHFileReaderEnabled(HoodieConfig hoodieConfig) {
return hoodieConfig.getBooleanOrDefault(HoodieReaderConfig.USE_NATIVE_HFILE_READER);
}
-}
+}
\ No newline at end of file
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileWriterFactory.java b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileWriterFactory.java
index c0e154ed6ab..65b172136c1 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileWriterFactory.java
+++ b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileWriterFactory.java
@@ -39,66 +39,71 @@ import static org.apache.hudi.common.model.HoodieFileFormat.ORC;
import static org.apache.hudi.common.model.HoodieFileFormat.PARQUET;
public class HoodieFileWriterFactory {
+ protected final StorageConfiguration<?> storageConf;
+
+ public HoodieFileWriterFactory(StorageConfiguration<?> storageConf) {
+ this.storageConf = storageConf;
+ }
public static <T, I, K, O> HoodieFileWriter getFileWriter(
String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier, HoodieRecordType recordType) throws IOException {
final String extension = FSUtils.getFileExtension(path.getName());
HoodieFileWriterFactory factory = HoodieIOFactory.getIOFactory(conf).getWriterFactory(recordType);
- return factory.getFileWriterByFormat(extension, instantTime, path, conf, config, schema, taskContextSupplier);
+ return factory.getFileWriterByFormat(extension, instantTime, path, config, schema, taskContextSupplier);
}
public static <T, I, K, O> HoodieFileWriter getFileWriter(HoodieFileFormat format, OutputStream outputStream,
StorageConfiguration<?> conf, HoodieConfig config, Schema schema, HoodieRecordType recordType)
throws IOException {
HoodieFileWriterFactory factory = HoodieIOFactory.getIOFactory(conf).getWriterFactory(recordType);
- return factory.getFileWriterByFormat(format, outputStream, conf, config, schema);
+ return factory.getFileWriterByFormat(format, outputStream, config, schema);
}
protected <T, I, K, O> HoodieFileWriter getFileWriterByFormat(
- String extension, String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
+ String extension, String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
if (PARQUET.getFileExtension().equals(extension)) {
- return newParquetFileWriter(instantTime, path, conf, config, schema, taskContextSupplier);
+ return newParquetFileWriter(instantTime, path, config, schema, taskContextSupplier);
}
if (HFILE.getFileExtension().equals(extension)) {
- return newHFileFileWriter(instantTime, path, conf, config, schema, taskContextSupplier);
+ return newHFileFileWriter(instantTime, path, config, schema, taskContextSupplier);
}
if (ORC.getFileExtension().equals(extension)) {
- return newOrcFileWriter(instantTime, path, conf, config, schema, taskContextSupplier);
+ return newOrcFileWriter(instantTime, path, config, schema, taskContextSupplier);
}
throw new UnsupportedOperationException(extension + " format not supported yet.");
}
protected <T, I, K, O> HoodieFileWriter getFileWriterByFormat(HoodieFileFormat format, OutputStream outputStream,
- StorageConfiguration<?> conf, HoodieConfig config, Schema schema) throws IOException {
+ HoodieConfig config, Schema schema) throws IOException {
switch (format) {
case PARQUET:
- return newParquetFileWriter(outputStream, conf, config, schema);
+ return newParquetFileWriter(outputStream, config, schema);
default:
throw new UnsupportedOperationException(format + " format not supported yet.");
}
}
protected HoodieFileWriter newParquetFileWriter(
- String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
+ String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
throw new UnsupportedOperationException();
}
protected HoodieFileWriter newParquetFileWriter(
- OutputStream outputStream, StorageConfiguration<?> conf, HoodieConfig config, Schema schema) throws IOException {
+ OutputStream outputStream, HoodieConfig config, Schema schema) throws IOException {
throw new UnsupportedOperationException();
}
protected HoodieFileWriter newHFileFileWriter(
- String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
+ String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
throw new UnsupportedOperationException();
}
protected HoodieFileWriter newOrcFileWriter(
- String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
+ String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
throw new UnsupportedOperationException();
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieIOFactory.java b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieIOFactory.java
index 3e715366134..3ae6b60321e 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieIOFactory.java
+++ b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieIOFactory.java
@@ -20,25 +20,30 @@
package org.apache.hudi.io.storage;
import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.fs.ConsistencyGuard;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.StoragePath;
/**
* Base class to get HoodieFileReaderFactory and HoodieFileWriterFactory
*/
public abstract class HoodieIOFactory {
+ protected final StorageConfiguration<?> storageConf;
+
+ public HoodieIOFactory(StorageConfiguration<?> storageConf) {
+ this.storageConf = storageConf;
+ }
public static HoodieIOFactory getIOFactory(StorageConfiguration<?> storageConf) {
String ioFactoryClass = storageConf.getString(HoodieStorageConfig.HOODIE_IO_FACTORY_CLASS.key())
.orElse(HoodieStorageConfig.HOODIE_IO_FACTORY_CLASS.defaultValue());
- return getIOFactory(ioFactoryClass);
- }
-
- private static HoodieIOFactory getIOFactory(String ioFactoryClass) {
try {
- return ReflectionUtils.loadClass(ioFactoryClass);
+ return (HoodieIOFactory) ReflectionUtils
+ .loadClass(ioFactoryClass, new Class<?>[] {StorageConfiguration.class}, storageConf);
} catch (Exception e) {
throw new HoodieException("Unable to create " + ioFactoryClass, e);
}
@@ -48,4 +53,13 @@ public abstract class HoodieIOFactory {
public abstract HoodieFileWriterFactory getWriterFactory(HoodieRecord.HoodieRecordType recordType);
+ public abstract HoodieStorage getStorage(StoragePath storagePath);
+
+ public abstract HoodieStorage getStorage(StoragePath path,
+ boolean enableRetry,
+ long maxRetryIntervalMs,
+ int maxRetryNumbers,
+ long initialRetryIntervalMs,
+ String retryExceptions,
+ ConsistencyGuard consistencyGuard);
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
index 74079e8845a..efdb1baf23d 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
@@ -447,7 +447,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
if (basefile.isPresent()) {
StoragePath baseFilePath = basefile.get().getStoragePath();
baseFileReader = (HoodieSeekingFileReader<?>) HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, getStorageConf(), baseFilePath);
+ .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, baseFilePath);
baseFileOpenMs = timer.endTimer();
LOG.info(String.format("Opened metadata base file from %s at instant %s in %d ms", baseFilePath,
basefile.get().getCommitTime(), baseFileOpenMs));
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 8c2ccf5f080..cf5e4b27dd7 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -505,8 +505,7 @@ public class HoodieTableMetadataUtil {
final StoragePath writeFilePath = new StoragePath(dataMetaClient.getBasePathV2(), pathWithPartition);
try (HoodieFileReader fileReader = HoodieIOFactory.getIOFactory(dataMetaClient.getStorageConf())
- .getReaderFactory(HoodieRecordType.AVRO).getFileReader(hoodieConfig,
- dataMetaClient.getStorageConf(), writeFilePath)) {
+ .getReaderFactory(HoodieRecordType.AVRO).getFileReader(hoodieConfig, writeFilePath)) {
try {
final BloomFilter fileBloomFilter = fileReader.readBloomFilter();
if (fileBloomFilter == null) {
@@ -927,7 +926,7 @@ public class HoodieTableMetadataUtil {
private static ByteBuffer readBloomFilter(StorageConfiguration<?> conf, StoragePath filePath) throws IOException {
HoodieConfig hoodieConfig = getReaderConfigs(conf);
try (HoodieFileReader fileReader = HoodieIOFactory.getIOFactory(conf).getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(hoodieConfig, conf, filePath)) {
+ .getFileReader(hoodieConfig, filePath)) {
final BloomFilter fileBloomFilter = fileReader.readBloomFilter();
if (fileBloomFilter == null) {
return null;
@@ -1782,7 +1781,7 @@ public class HoodieTableMetadataUtil {
final String fileId = baseFile.getFileId();
final String instantTime = baseFile.getCommitTime();
HoodieFileReader reader = HoodieIOFactory.getIOFactory(configuration).getReaderFactory(HoodieRecord.HoodieRecordType.AVRO)
- .getFileReader(config, configuration, dataFilePath);
+ .getFileReader(config, dataFilePath);
return getHoodieRecordIterator(reader.getRecordKeyIterator(), forDelete, partition, fileId, instantTime);
});
}
@@ -1843,7 +1842,7 @@ public class HoodieTableMetadataUtil {
final String instantTime = baseFile.getCommitTime();
HoodieConfig hoodieConfig = getReaderConfigs(storageConf);
HoodieFileReader reader = HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(HoodieRecord.HoodieRecordType.AVRO)
- .getFileReader(hoodieConfig, storageConf, dataFilePath);
+ .getFileReader(hoodieConfig, dataFilePath);
return getHoodieRecordIterator(reader.getRecordKeyIterator(), forDelete, partition, fileId, instantTime);
});
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/storage/HoodieStorageUtils.java b/hudi-common/src/main/java/org/apache/hudi/storage/HoodieStorageUtils.java
index 64bcde90d71..770fc77372e 100644
--- a/hudi-common/src/main/java/org/apache/hudi/storage/HoodieStorageUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/storage/HoodieStorageUtils.java
@@ -19,63 +19,20 @@
package org.apache.hudi.storage;
-import org.apache.hudi.common.fs.ConsistencyGuard;
-import org.apache.hudi.common.util.ReflectionUtils;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
+import static org.apache.hudi.io.storage.HoodieIOFactory.getIOFactory;
public class HoodieStorageUtils {
- public static final String HUDI_HADOOP_STORAGE = "org.apache.hudi.storage.hadoop.HoodieHadoopStorage";
- public static final String HADOOP_STORAGE_CONF = "org.apache.hudi.storage.hadoop.HadoopStorageConfiguration";
public static final String DEFAULT_URI = "file:///";
public static HoodieStorage getStorage(StorageConfiguration<?> conf) {
return getStorage(DEFAULT_URI, conf);
}
- public static HoodieStorage getStorage(FileSystem fs) {
- return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE, new Class<?>[] {FileSystem.class}, fs);
- }
-
public static HoodieStorage getStorage(String basePath, StorageConfiguration<?> conf) {
- return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE, new Class<?>[] {String.class, StorageConfiguration.class}, basePath, conf);
- }
-
- public static HoodieStorage getStorage(String basePath, Configuration conf) {
- return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE, new Class<?>[] {String.class, Configuration.class}, basePath, conf);
+ return getStorage(new StoragePath(basePath), conf);
}
public static HoodieStorage getStorage(StoragePath path, StorageConfiguration<?> conf) {
- return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE, new Class<?>[] {StoragePath.class, StorageConfiguration.class}, path, conf);
- }
-
- public static HoodieStorage getStorage(StoragePath path,
- StorageConfiguration<?> conf,
- boolean enableRetry,
- long maxRetryIntervalMs,
- int maxRetryNumbers,
- long initialRetryIntervalMs,
- String retryExceptions,
- ConsistencyGuard consistencyGuard) {
- return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE,
- new Class<?>[] {StoragePath.class, StorageConfiguration.class, boolean.class, long.class, int.class, long.class,
- String.class, ConsistencyGuard.class},
- path, conf, enableRetry, maxRetryIntervalMs, maxRetryNumbers, initialRetryIntervalMs, retryExceptions,
- consistencyGuard);
- }
-
- public static HoodieStorage getRawStorage(HoodieStorage storage) {
- return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE, new Class<?>[] {HoodieStorage.class}, storage);
- }
-
- public static StorageConfiguration<?> getStorageConf(Configuration conf) {
- return (StorageConfiguration<?>) ReflectionUtils.loadClass(HADOOP_STORAGE_CONF,
- new Class<?>[] {Configuration.class}, conf);
- }
-
- public static StorageConfiguration<Configuration> getStorageConfWithCopy(Configuration conf) {
- return (StorageConfiguration<Configuration>) ReflectionUtils.loadClass(HADOOP_STORAGE_CONF,
- new Class<?>[] {Configuration.class, boolean.class}, conf, true);
+ return getIOFactory(conf).getStorage(path);
}
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
index d0af0ae8963..074d9b1c020 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
@@ -51,8 +51,6 @@ import java.util.Objects;
import java.util.Properties;
import java.util.UUID;
-import static org.apache.hudi.storage.HoodieStorageUtils.HADOOP_STORAGE_CONF;
-
/**
* A utility class for testing.
*/
@@ -63,6 +61,7 @@ public class HoodieTestUtils {
public static final String DEFAULT_WRITE_TOKEN = "1-0-1";
public static final int DEFAULT_LOG_VERSION = 1;
public static final String[] DEFAULT_PARTITION_PATHS = {"2016/03/15", "2015/03/16", "2015/03/17"};
+ public static final String HADOOP_STORAGE_CONF = "org.apache.hudi.storage.hadoop.HadoopStorageConfiguration";
public static StorageConfiguration<Configuration> getDefaultStorageConf() {
return (StorageConfiguration<Configuration>) ReflectionUtils.loadClass(HADOOP_STORAGE_CONF,
@@ -211,16 +210,6 @@ public class HoodieTestUtils {
.setConf(storageConf).setBasePath(basePath).build();
}
- /**
- * @param conf file system configuration.
- * @param basePath base path of the Hudi table.
- * @return a new {@link HoodieTableMetaClient} instance.
- */
- public static HoodieTableMetaClient createMetaClient(Configuration conf,
- String basePath) {
- return createMetaClient(HoodieStorageUtils.getStorageConfWithCopy(conf), basePath);
- }
-
/**
* @param storage {@link HoodieStorage} instance.
* @param basePath base path of the Hudi table.
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java
index 3709c27a8b8..460e36154cf 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java
@@ -276,7 +276,7 @@ public class ClusteringOperator extends TableStreamOperator<ClusteringCommitEven
? Option.empty()
: Option.of(HoodieIOFactory.getIOFactory(table.getStorageConf())
.getReaderFactory(table.getConfig().getRecordMerger().getRecordType())
- .getFileReader(table.getConfig(), table.getStorageConf(), new StoragePath(clusteringOp.getDataFilePath())));
+ .getFileReader(table.getConfig(), new StoragePath(clusteringOp.getDataFilePath())));
HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
.withStorage(table.getMetaClient().getStorage())
.withBasePath(table.getMetaClient().getBasePath())
@@ -325,7 +325,7 @@ public class ClusteringOperator extends TableStreamOperator<ClusteringCommitEven
HoodieFileReaderFactory fileReaderFactory = HoodieIOFactory.getIOFactory(table.getStorageConf())
.getReaderFactory(table.getConfig().getRecordMerger().getRecordType());
HoodieAvroFileReader fileReader = (HoodieAvroFileReader) fileReaderFactory.getFileReader(
- table.getConfig(), table.getStorageConf(), new StoragePath(clusteringOp.getDataFilePath()));
+ table.getConfig(), new StoragePath(clusteringOp.getDataFilePath()));
return new CloseableMappingIterator<>(fileReader.getRecordIterator(readerSchema), HoodieRecord::getData);
} catch (IOException e) {
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
index c612d1f1365..6ecf1b33045 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
@@ -38,6 +38,7 @@ import org.apache.hudi.sink.event.WriteMetadataEvent;
import org.apache.hudi.sink.utils.MockCoordinatorExecutor;
import org.apache.hudi.sink.utils.NonThrownExecutor;
import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.utils.TestConfigurations;
import org.apache.hudi.utils.TestUtils;
@@ -314,7 +315,7 @@ public class TestStreamWriteOperatorCoordinator {
assertNotEquals("", instant);
final String metadataTableBasePath = HoodieTableMetadata.getMetadataTableBasePath(tempFile.getAbsolutePath());
- HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils.createMetaClient(HadoopConfigurations.getHadoopConf(conf), metadataTableBasePath);
+ HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(conf)), metadataTableBasePath);
HoodieTimeline completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(1));
assertThat(completedTimeline.lastInstant().get().getTimestamp(), startsWith(HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP));
@@ -386,7 +387,7 @@ public class TestStreamWriteOperatorCoordinator {
assertNotEquals("", instant);
final String metadataTableBasePath = HoodieTableMetadata.getMetadataTableBasePath(tempFile.getAbsolutePath());
- HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils.createMetaClient(HadoopConfigurations.getHadoopConf(conf), metadataTableBasePath);
+ HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(conf)), metadataTableBasePath);
HoodieTimeline completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(1));
assertThat(completedTimeline.lastInstant().get().getTimestamp(), startsWith(HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP));
@@ -431,7 +432,7 @@ public class TestStreamWriteOperatorCoordinator {
assertNotEquals("", instant);
final String metadataTableBasePath = HoodieTableMetadata.getMetadataTableBasePath(tempFile.getAbsolutePath());
- HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils.createMetaClient(HadoopConfigurations.getHadoopConf(conf), metadataTableBasePath);
+ HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(conf)), metadataTableBasePath);
HoodieTimeline completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(1));
assertThat(completedTimeline.lastInstant().get().getTimestamp(), startsWith(HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP));
@@ -514,7 +515,7 @@ public class TestStreamWriteOperatorCoordinator {
assertNotEquals("", instant);
final String metadataTableBasePath = HoodieTableMetadata.getMetadataTableBasePath(tempFile.getAbsolutePath());
- HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils.createMetaClient(HadoopConfigurations.getHadoopConf(conf), metadataTableBasePath);
+ HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(conf)), metadataTableBasePath);
HoodieTimeline completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(1));
assertThat(completedTimeline.lastInstant().get().getTimestamp(), startsWith(HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP));
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/HoodieCatalogTestUtils.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/HoodieCatalogTestUtils.java
index c98b4ac0da2..bf54fe27009 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/HoodieCatalogTestUtils.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/HoodieCatalogTestUtils.java
@@ -18,6 +18,9 @@
package org.apache.hudi.table.catalog;
+import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.hadoop.hive.conf.HiveConf;
@@ -74,4 +77,8 @@ public class HoodieCatalogTestUtils {
throw new CatalogException("Failed to create test HiveConf to HiveCatalog.", e);
}
}
+
+ public static StorageConfiguration<?> createStorageConf() {
+ return new HadoopStorageConfiguration(createHiveConf());
+ }
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java
index 2781e3f8153..98c98bebcce 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java
@@ -33,6 +33,7 @@ import org.apache.hudi.keygen.ComplexAvroKeyGenerator;
import org.apache.hudi.keygen.NonpartitionedAvroKeyGenerator;
import org.apache.hudi.keygen.SimpleAvroKeyGenerator;
import org.apache.hudi.sink.partitioner.profile.WriteProfiles;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.util.StreamerUtil;
import org.apache.hudi.utils.TestConfigurations;
import org.apache.hudi.utils.TestData;
@@ -424,7 +425,8 @@ public class TestHoodieCatalog {
String tablePathStr = catalog.inferTablePath(catalogPathStr, tablePath);
Configuration flinkConf = TestConfigurations.getDefaultConf(tablePathStr);
- HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(HadoopConfigurations.getHadoopConf(flinkConf), tablePathStr);
+ HoodieTableMetaClient metaClient = HoodieTestUtils
+ .createMetaClient(new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(flinkConf)), tablePathStr);
TestData.writeData(TestData.DATA_SET_INSERT, flinkConf);
assertTrue(catalog.partitionExists(tablePath, partitionSpec));
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java
index 22755d339d4..fde58caa5e4 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java
@@ -35,6 +35,7 @@ import org.apache.hudi.keygen.NonpartitionedAvroKeyGenerator;
import org.apache.hudi.keygen.SimpleAvroKeyGenerator;
import org.apache.hudi.sink.partitioner.profile.WriteProfiles;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.util.StreamerUtil;
import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
@@ -75,7 +76,7 @@ import java.util.stream.Collectors;
import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR;
import static org.apache.hudi.configuration.FlinkOptions.PRECOMBINE_FIELD;
import static org.apache.hudi.keygen.constant.KeyGeneratorOptions.RECORDKEY_FIELD_NAME;
-import static org.apache.hudi.table.catalog.HoodieCatalogTestUtils.createHiveConf;
+import static org.apache.hudi.table.catalog.HoodieCatalogTestUtils.createStorageConf;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -221,7 +222,7 @@ public class TestHoodieHiveCatalog {
// validate key generator for partitioned table
HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- createHiveConf(), hoodieCatalog.inferTablePath(tablePath, table));
+ createStorageConf(), hoodieCatalog.inferTablePath(tablePath, table));
String keyGeneratorClassName = metaClient.getTableConfig().getKeyGeneratorClassName();
assertEquals(keyGeneratorClassName, SimpleAvroKeyGenerator.class.getName());
@@ -232,7 +233,7 @@ public class TestHoodieHiveCatalog {
hoodieCatalog.createTable(singleKeyMultiPartitionPath, singleKeyMultiPartitionTable, false);
HoodieTableMetaClient singleKeyMultiPartitionTableMetaClient = HoodieTestUtils.createMetaClient(
- createHiveConf(),
+ createStorageConf(),
hoodieCatalog.inferTablePath(singleKeyMultiPartitionPath, singleKeyMultiPartitionTable));
assertThat(singleKeyMultiPartitionTableMetaClient.getTableConfig().getKeyGeneratorClassName(), is(ComplexAvroKeyGenerator.class.getName()));
@@ -245,7 +246,7 @@ public class TestHoodieHiveCatalog {
hoodieCatalog.createTable(multiKeySinglePartitionPath, multiKeySinglePartitionTable, false);
HoodieTableMetaClient multiKeySinglePartitionTableMetaClient = HoodieTestUtils.createMetaClient(
- createHiveConf(),
+ createStorageConf(),
hoodieCatalog.inferTablePath(multiKeySinglePartitionPath, multiKeySinglePartitionTable));
assertThat(multiKeySinglePartitionTableMetaClient.getTableConfig().getKeyGeneratorClassName(), is(ComplexAvroKeyGenerator.class.getName()));
@@ -256,7 +257,7 @@ public class TestHoodieHiveCatalog {
hoodieCatalog.createTable(nonPartitionPath, nonPartitionTable, false);
metaClient = HoodieTestUtils.createMetaClient(
- createHiveConf(), hoodieCatalog.inferTablePath(nonPartitionPath, nonPartitionTable));
+ createStorageConf(), hoodieCatalog.inferTablePath(nonPartitionPath, nonPartitionTable));
keyGeneratorClassName = metaClient.getTableConfig().getKeyGeneratorClassName();
assertEquals(keyGeneratorClassName, NonpartitionedAvroKeyGenerator.class.getName());
}
@@ -325,7 +326,7 @@ public class TestHoodieHiveCatalog {
hoodieCatalog.createTable(tablePath, table, true);
HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- createHiveConf(), hoodieCatalog.inferTablePath(tablePath, table));
+ createStorageConf(), hoodieCatalog.inferTablePath(tablePath, table));
return metaClient.getTableConfig().getProps();
}
@@ -450,7 +451,7 @@ public class TestHoodieHiveCatalog {
hoodieCatalog.dropPartition(tablePath, partitionSpec, false);
String tablePathStr = hoodieCatalog.inferTablePath(tablePath, hoodieCatalog.getTable(tablePath));
- HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(hoodieCatalog.getHiveConf(), tablePathStr);
+ HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(hoodieCatalog.getHiveConf()), tablePathStr);
HoodieInstant latestInstant = metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().orElse(null);
assertNotNull(latestInstant, "Delete partition commit should be completed");
HoodieCommitMetadata commitMetadata = WriteProfiles.getCommitMetadata(tablePath.getObjectName(), new org.apache.flink.core.fs.Path(tablePathStr),
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java
index 1999791ab30..5cb7f02054f 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java
@@ -32,6 +32,7 @@ import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.HadoopConfigurations;
import org.apache.hudi.source.IncrementalInputSplits;
import org.apache.hudi.source.prune.PartitionPruners;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.table.HoodieTableSource;
import org.apache.hudi.table.format.cdc.CdcInputFormat;
import org.apache.hudi.table.format.cow.CopyOnWriteInputFormat;
@@ -778,7 +779,7 @@ public class TestInputFormat {
}
HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(conf), tempFile.getAbsolutePath());
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(conf)), tempFile.getAbsolutePath());
List<String> commits = metaClient.getCommitsTimeline().filterCompletedInstants().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).collect(Collectors.toList());
@@ -863,7 +864,7 @@ public class TestInputFormat {
}
HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(conf), tempFile.getAbsolutePath());
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(conf)), tempFile.getAbsolutePath());
List<String> commits = metaClient.getCommitsTimeline().filterCompletedInstants().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).collect(Collectors.toList());
@@ -1013,7 +1014,7 @@ public class TestInputFormat {
writeClient.clean();
HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(conf), tempFile.getAbsolutePath());
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(conf)), tempFile.getAbsolutePath());
List<String> commits = metaClient.getCommitsTimeline().filterCompletedInstants().getInstantsAsStream()
.map(HoodieInstant::getTimestamp).collect(Collectors.toList());
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestUtils.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestUtils.java
index 0ccf9f9b75a..6f495a0ab7f 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestUtils.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestUtils.java
@@ -31,6 +31,7 @@ import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.HadoopConfigurations;
import org.apache.hudi.source.StreamReadMonitoringFunction;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.table.format.mor.MergeOnReadInputSplit;
import org.apache.hudi.util.StreamerUtil;
@@ -48,19 +49,19 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestUtils {
public static String getLastPendingInstant(String basePath) {
final HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(new Configuration()), basePath);
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())), basePath);
return StreamerUtil.getLastPendingInstant(metaClient);
}
public static String getLastCompleteInstant(String basePath) {
final HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(new Configuration()), basePath);
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())), basePath);
return StreamerUtil.getLastCompletedInstant(metaClient);
}
public static String getLastCompleteInstant(String basePath, String commitAction) {
final HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(new Configuration()), basePath);
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())), basePath);
return metaClient.getCommitsTimeline().filterCompletedInstants()
.filter(instant -> commitAction.equals(instant.getAction()))
.lastInstant()
@@ -70,7 +71,7 @@ public class TestUtils {
public static String getLastDeltaCompleteInstant(String basePath) {
final HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(new Configuration()), basePath);
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())), basePath);
return metaClient.getCommitsTimeline().filterCompletedInstants()
.filter(hoodieInstant -> hoodieInstant.getAction().equals(HoodieTimeline.DELTA_COMMIT_ACTION))
.lastInstant()
@@ -80,7 +81,7 @@ public class TestUtils {
public static String getFirstCompleteInstant(String basePath) {
final HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(new Configuration()), basePath);
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())), basePath);
return metaClient.getCommitsAndCompactionTimeline().filterCompletedInstants().firstInstant()
.map(HoodieInstant::getTimestamp).orElse(null);
}
@@ -88,7 +89,7 @@ public class TestUtils {
@Nullable
public static String getNthCompleteInstant(String basePath, int n, String action) {
final HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(new Configuration()), basePath);
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())), basePath);
return metaClient.getActiveTimeline()
.filterCompletedInstants()
.filter(instant -> action.equals(instant.getAction()))
@@ -99,7 +100,7 @@ public class TestUtils {
@Nullable
public static String getNthArchivedInstant(String basePath, int n) {
final HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(new Configuration()), basePath);
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())), basePath);
return metaClient.getArchivedTimeline().getCommitsTimeline().filterCompletedInstants()
.nthInstant(n).map(HoodieInstant::getTimestamp).orElse(null);
}
@@ -118,7 +119,7 @@ public class TestUtils {
public static int getCompletedInstantCount(String basePath, String action) {
final HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- HadoopConfigurations.getHadoopConf(new Configuration()), basePath);
+ new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())), basePath);
return metaClient.getActiveTimeline()
.filterCompletedInstants()
.filter(instant -> action.equals(instant.getAction()))
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/HFileUtils.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/HFileUtils.java
index 52c26477f47..119c0ed5aec 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/HFileUtils.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/HFileUtils.java
@@ -104,7 +104,6 @@ public class HFileUtils extends BaseFileUtils {
.getReaderFactory(HoodieRecord.HoodieRecordType.AVRO)
.getFileReader(
ConfigUtils.DEFAULT_HUDI_CONFIG_FOR_READER,
- configuration,
filePath)) {
return fileReader.getSchema();
} catch (IOException e) {
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
index 44be55438a1..3cbdd6a4949 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
@@ -30,7 +30,6 @@ import org.apache.hudi.common.util.collection.ImmutablePair;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.InvalidHoodiePathException;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
@@ -453,7 +452,7 @@ public class HadoopFSUtils {
List<String> subPaths) {
Map<String, T> result = new HashMap<>();
if (subPaths.size() > 0) {
- StorageConfiguration<Configuration> conf = HoodieStorageUtils.getStorageConfWithCopy(fs.getConf());
+ StorageConfiguration<Configuration> conf = new HadoopStorageConfiguration(fs.getConf(), true);
int actualParallelism = Math.min(subPaths.size(), parallelism);
hoodieEngineContext.setJobStatus(FSUtils.class.getSimpleName(),
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 3903d95b9d9..d3a340adfbb 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
@@ -21,8 +21,6 @@ package org.apache.hudi.io.hadoop;
import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.ReflectionUtils;
-import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.io.storage.HoodieAvroBootstrapFileReader;
import org.apache.hudi.io.storage.HoodieFileReader;
import org.apache.hudi.io.storage.HoodieFileReaderFactory;
@@ -36,60 +34,48 @@ import org.apache.avro.Schema;
import java.io.IOException;
public class HoodieAvroFileReaderFactory extends HoodieFileReaderFactory {
- public static final String HBASE_AVRO_HFILE_READER = "org.apache.hudi.io.hadoop.HoodieHBaseAvroHFileReader";
+
+ public HoodieAvroFileReaderFactory(StorageConfiguration<?> storageConf) {
+ super(storageConf);
+ }
@Override
- protected HoodieFileReader newParquetFileReader(StorageConfiguration<?> conf, StoragePath path) {
- return new HoodieAvroParquetReader(conf, path);
+ protected HoodieFileReader newParquetFileReader(StoragePath path) {
+ return new HoodieAvroParquetReader(storageConf, path);
}
@Override
protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
- StorageConfiguration<?> conf,
StoragePath path,
Option<Schema> schemaOption) throws IOException {
if (isUseNativeHFileReaderEnabled(hoodieConfig)) {
- return new HoodieNativeAvroHFileReader(conf, path, schemaOption);
+ return new HoodieNativeAvroHFileReader(storageConf, path, schemaOption);
}
- try {
- if (schemaOption.isPresent()) {
- return (HoodieFileReader) ReflectionUtils.loadClass(HBASE_AVRO_HFILE_READER,
- new Class<?>[] {StorageConfiguration.class, StoragePath.class, Option.class}, conf, path, schemaOption);
- }
- return (HoodieFileReader) ReflectionUtils.loadClass(HBASE_AVRO_HFILE_READER,
- new Class<?>[] {StorageConfiguration.class, StoragePath.class}, conf, path);
- } catch (HoodieException e) {
- throw new IOException("Cannot instantiate HoodieHBaseAvroHFileReader", e);
+ if (schemaOption.isPresent()) {
+ return new HoodieHBaseAvroHFileReader(storageConf, path, schemaOption);
}
+ return new HoodieHBaseAvroHFileReader(storageConf, path);
}
@Override
protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
- StorageConfiguration<?> conf,
StoragePath path,
HoodieStorage storage,
byte[] content,
- Option<Schema> schemaOption)
- throws IOException {
+ Option<Schema> schemaOption) throws IOException {
if (isUseNativeHFileReaderEnabled(hoodieConfig)) {
- return new HoodieNativeAvroHFileReader(conf, content, schemaOption);
- }
- try {
- return (HoodieFileReader) ReflectionUtils.loadClass(HBASE_AVRO_HFILE_READER,
- new Class<?>[] {StorageConfiguration.class, StoragePath.class, HoodieStorage.class, byte[].class, Option.class},
- conf, path, storage, content, schemaOption);
- } catch (HoodieException e) {
- throw new IOException("Cannot instantiate HoodieHBaseAvroHFileReader", e);
+ return new HoodieNativeAvroHFileReader(storageConf, content, schemaOption);
}
+ return new HoodieHBaseAvroHFileReader(storageConf, path, storage, content, schemaOption);
}
@Override
- protected HoodieFileReader newOrcFileReader(StorageConfiguration<?> conf, StoragePath path) {
- return new HoodieAvroOrcReader(conf, path);
+ protected HoodieFileReader newOrcFileReader(StoragePath path) {
+ return new HoodieAvroOrcReader(storageConf, path);
}
@Override
public HoodieFileReader newBootstrapFileReader(HoodieFileReader skeletonFileReader, HoodieFileReader dataFileReader, Option<String[]> partitionFields, Object[] partitionValues) {
return new HoodieAvroBootstrapFileReader(skeletonFileReader, dataFileReader, partitionFields, partitionValues);
}
-}
+}
\ No newline at end of file
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieAvroFileWriterFactory.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieAvroFileWriterFactory.java
index d0b8faa7589..0ce60074c2d 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieAvroFileWriterFactory.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieAvroFileWriterFactory.java
@@ -27,7 +27,6 @@ import org.apache.hudi.common.engine.TaskContextSupplier;
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.exception.HoodieException;
import org.apache.hudi.io.storage.HoodieAvroHFileReaderImplBase;
import org.apache.hudi.io.storage.HoodieFileWriter;
import org.apache.hudi.io.storage.HoodieFileWriterFactory;
@@ -55,17 +54,17 @@ 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 {
- //hardcoded classes to remove at a later time
- public static final String HOODIE_AVRO_PARQUET_WRITER = "org.apache.hudi.io.hadoop.HoodieAvroParquetWriter";
- public static final String HOODIE_AVRO_HFILE_WRITER = "org.apache.hudi.io.hadoop.HoodieAvroHFileWriter";
- public static final String HOODIE_AVRO_ORC_WRITER = "org.apache.hudi.io.hadoop.HoodieAvroOrcWriter";
+
+ public HoodieAvroFileWriterFactory(StorageConfiguration<?> storageConf) {
+ super(storageConf);
+ }
@Override
protected HoodieFileWriter newParquetFileWriter(
- String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
+ String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
boolean populateMetaFields = config.getBooleanOrDefault(HoodieTableConfig.POPULATE_META_FIELDS);
- HoodieAvroWriteSupport writeSupport = getHoodieAvroWriteSupport(conf, schema, config, enableBloomFilter(populateMetaFields, config));
+ HoodieAvroWriteSupport writeSupport = getHoodieAvroWriteSupport(schema, config, enableBloomFilter(populateMetaFields, config));
String compressionCodecName = config.getStringOrDefault(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME);
// Support PARQUET_COMPRESSION_CODEC_NAME is ""
@@ -77,76 +76,56 @@ public class HoodieAvroFileWriterFactory extends HoodieFileWriterFactory {
config.getIntOrDefault(HoodieStorageConfig.PARQUET_BLOCK_SIZE),
config.getIntOrDefault(HoodieStorageConfig.PARQUET_PAGE_SIZE),
config.getLongOrDefault(HoodieStorageConfig.PARQUET_MAX_FILE_SIZE),
- conf, config.getDoubleOrDefault(HoodieStorageConfig.PARQUET_COMPRESSION_RATIO_FRACTION),
+ storageConf, config.getDoubleOrDefault(HoodieStorageConfig.PARQUET_COMPRESSION_RATIO_FRACTION),
config.getBooleanOrDefault(HoodieStorageConfig.PARQUET_DICTIONARY_ENABLED));
- try {
- return (HoodieFileWriter) ReflectionUtils.loadClass(HOODIE_AVRO_PARQUET_WRITER,
- new Class<?>[] {StoragePath.class, HoodieParquetConfig.class, String.class, TaskContextSupplier.class, boolean.class},
- path, parquetConfig, instantTime, taskContextSupplier, populateMetaFields);
- } catch (HoodieException e) {
- throw (IOException) e.getCause().getCause();
- }
-
+ return new HoodieAvroParquetWriter(path, parquetConfig, instantTime, taskContextSupplier, populateMetaFields);
}
protected HoodieFileWriter newParquetFileWriter(
- OutputStream outputStream, StorageConfiguration<?> conf, HoodieConfig config, Schema schema) throws IOException {
- HoodieAvroWriteSupport writeSupport = getHoodieAvroWriteSupport(conf, schema, config, false);
+ OutputStream outputStream, HoodieConfig config, Schema schema) throws IOException {
+ HoodieAvroWriteSupport writeSupport = getHoodieAvroWriteSupport(schema, config, false);
HoodieParquetConfig<HoodieAvroWriteSupport> parquetConfig = new HoodieParquetConfig<>(writeSupport,
CompressionCodecName.fromConf(config.getString(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME)),
config.getInt(HoodieStorageConfig.PARQUET_BLOCK_SIZE),
config.getInt(HoodieStorageConfig.PARQUET_PAGE_SIZE),
config.getLong(HoodieStorageConfig.PARQUET_MAX_FILE_SIZE), // todo: 1024*1024*1024
- conf, config.getDouble(HoodieStorageConfig.PARQUET_COMPRESSION_RATIO_FRACTION),
+ storageConf, config.getDouble(HoodieStorageConfig.PARQUET_COMPRESSION_RATIO_FRACTION),
config.getBoolean(HoodieStorageConfig.PARQUET_DICTIONARY_ENABLED));
return new HoodieParquetStreamWriter(new FSDataOutputStream(outputStream, null), parquetConfig);
}
protected HoodieFileWriter newHFileFileWriter(
- String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
+ String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
BloomFilter filter = createBloomFilter(config);
- HoodieHFileConfig hfileConfig = new HoodieHFileConfig(conf.unwrapAs(Configuration.class),
+ HoodieHFileConfig hfileConfig = new HoodieHFileConfig(storageConf.unwrapAs(Configuration.class),
Compression.Algorithm.valueOf(
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);
-
- try {
- return (HoodieFileWriter) ReflectionUtils.loadClass(HOODIE_AVRO_HFILE_WRITER,
- new Class<?>[] {String.class, StoragePath.class, HoodieHFileConfig.class, Schema.class, TaskContextSupplier.class, boolean.class},
- instantTime, path, hfileConfig, schema, taskContextSupplier, config.getBoolean(HoodieTableConfig.POPULATE_META_FIELDS));
- } catch (HoodieException e) {
- throw (IOException) e.getCause().getCause();
- }
+ return new HoodieAvroHFileWriter(instantTime, path, hfileConfig, schema, taskContextSupplier, config.getBoolean(HoodieTableConfig.POPULATE_META_FIELDS));
}
protected HoodieFileWriter newOrcFileWriter(
- String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
+ String instantTime, StoragePath path, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
BloomFilter filter = createBloomFilter(config);
- HoodieOrcConfig orcConfig = new HoodieOrcConfig(conf,
+ HoodieOrcConfig orcConfig = new HoodieOrcConfig(storageConf,
CompressionKind.valueOf(config.getString(HoodieStorageConfig.ORC_COMPRESSION_CODEC_NAME)),
config.getInt(HoodieStorageConfig.ORC_STRIPE_SIZE),
config.getInt(HoodieStorageConfig.ORC_BLOCK_SIZE),
config.getLong(HoodieStorageConfig.ORC_FILE_MAX_SIZE), filter);
- try {
- return (HoodieFileWriter) ReflectionUtils.loadClass(HOODIE_AVRO_ORC_WRITER,
- new Class<?>[] {String.class, StoragePath.class, HoodieOrcConfig.class, Schema.class, TaskContextSupplier.class},
- instantTime, path, orcConfig, schema, taskContextSupplier);
- } catch (HoodieException e) {
- throw (IOException) e.getCause().getCause();
- }
+ return new HoodieAvroOrcWriter(instantTime, path, orcConfig, schema, taskContextSupplier);
}
- private HoodieAvroWriteSupport getHoodieAvroWriteSupport(StorageConfiguration<?> conf, Schema schema,
+ private HoodieAvroWriteSupport getHoodieAvroWriteSupport(Schema schema,
HoodieConfig config, boolean enableBloomFilter) {
Option<BloomFilter> filter = enableBloomFilter ? Option.of(createBloomFilter(config)) : Option.empty();
return (HoodieAvroWriteSupport) ReflectionUtils.loadClass(
config.getStringOrDefault(HoodieStorageConfig.HOODIE_AVRO_WRITE_SUPPORT_CLASS),
new Class<?>[] {MessageType.class, Schema.class, Option.class, Properties.class},
- new AvroSchemaConverter(conf.unwrapAs(Configuration.class)).convert(schema), schema, filter, config.getProps());
+ new AvroSchemaConverter(storageConf.unwrapAs(Configuration.class)).convert(schema), schema, filter, config.getProps());
}
-}
+}
\ No newline at end of file
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieHadoopIOFactory.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieHadoopIOFactory.java
index 65c8d028adb..c357a70be3e 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieHadoopIOFactory.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieHadoopIOFactory.java
@@ -19,11 +19,16 @@
package org.apache.hudi.io.storage;
+import org.apache.hudi.common.fs.ConsistencyGuard;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.io.hadoop.HoodieAvroFileReaderFactory;
import org.apache.hudi.io.hadoop.HoodieAvroFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
/**
* Creates readers and writers for AVRO record payloads.
@@ -32,15 +37,21 @@ import org.apache.hudi.io.hadoop.HoodieAvroFileWriterFactory;
*/
public class HoodieHadoopIOFactory extends HoodieIOFactory {
+ public HoodieHadoopIOFactory(StorageConfiguration<?> storageConf) {
+ super(storageConf);
+ }
+
@Override
public HoodieFileReaderFactory getReaderFactory(HoodieRecord.HoodieRecordType recordType) {
switch (recordType) {
case AVRO:
- return new HoodieAvroFileReaderFactory();
+ return new HoodieAvroFileReaderFactory(storageConf);
case SPARK:
//TODO: remove this case [HUDI-7746]
try {
- return ReflectionUtils.loadClass("org.apache.hudi.io.storage.HoodieSparkFileReaderFactory");
+ return (HoodieFileReaderFactory) ReflectionUtils
+ .loadClass("org.apache.hudi.io.storage.HoodieSparkFileReaderFactory",
+ new Class<?>[] {StorageConfiguration.class}, storageConf);
} catch (Exception e) {
throw new HoodieException("Unable to create HoodieSparkFileReaderFactory", e);
}
@@ -53,11 +64,13 @@ public class HoodieHadoopIOFactory extends HoodieIOFactory {
public HoodieFileWriterFactory getWriterFactory(HoodieRecord.HoodieRecordType recordType) {
switch (recordType) {
case AVRO:
- return new HoodieAvroFileWriterFactory();
+ return new HoodieAvroFileWriterFactory(storageConf);
case SPARK:
//TODO: remove this case [HUDI-7746]
try {
- return ReflectionUtils.loadClass("org.apache.hudi.io.storage.HoodieSparkFileWriterFactory");
+ return (HoodieFileWriterFactory) ReflectionUtils
+ .loadClass("org.apache.hudi.io.storage.HoodieSparkFileWriterFactory",
+ new Class<?>[] {StorageConfiguration.class}, storageConf);
} catch (Exception e) {
throw new HoodieException("Unable to create HoodieSparkFileWriterFactory", e);
}
@@ -65,4 +78,21 @@ public class HoodieHadoopIOFactory extends HoodieIOFactory {
throw new UnsupportedOperationException(recordType + " record type not supported");
}
}
-}
+
+ @Override
+ public HoodieStorage getStorage(StoragePath storagePath) {
+ return new HoodieHadoopStorage(storagePath, storageConf);
+ }
+
+ @Override
+ public HoodieStorage getStorage(StoragePath path,
+ boolean enableRetry,
+ long maxRetryIntervalMs,
+ int maxRetryNumbers,
+ long initialRetryIntervalMs,
+ String retryExceptions,
+ ConsistencyGuard consistencyGuard) {
+ return new HoodieHadoopStorage(path, storageConf, enableRetry, maxRetryIntervalMs,
+ maxRetryNumbers, maxRetryIntervalMs, retryExceptions, consistencyGuard);
+ }
+}
\ No newline at end of file
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java
index 126b17617eb..72262f6b5d4 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java
@@ -58,27 +58,10 @@ import static org.apache.hudi.hadoop.fs.HadoopFSUtils.getFs;
public class HoodieHadoopStorage extends HoodieStorage {
private final FileSystem fs;
- public HoodieHadoopStorage(HoodieStorage storage) {
- FileSystem fs = (FileSystem) storage.getFileSystem();
- if (fs instanceof HoodieWrapperFileSystem) {
- this.fs = ((HoodieWrapperFileSystem) fs).getFileSystem();
- } else {
- this.fs = fs;
- }
- }
-
- public HoodieHadoopStorage(String basePath, Configuration conf) {
- this(HadoopFSUtils.getFs(basePath, conf));
- }
-
public HoodieHadoopStorage(StoragePath path, StorageConfiguration<?> conf) {
this(HadoopFSUtils.getFs(path, conf.unwrapAs(Configuration.class)));
}
- public HoodieHadoopStorage(String basePath, StorageConfiguration<?> conf) {
- this(HadoopFSUtils.getFs(basePath, conf));
- }
-
public HoodieHadoopStorage(StoragePath path,
StorageConfiguration<?> conf,
boolean enableRetry,
@@ -258,6 +241,15 @@ public class HoodieHadoopStorage extends HoodieStorage {
return fs.getConf();
}
+ @Override
+ public HoodieStorage getRawStorage() {
+ if (fs instanceof HoodieWrapperFileSystem) {
+ return new HoodieHadoopStorage(((HoodieWrapperFileSystem) fs).getFileSystem());
+ } else {
+ return this;
+ }
+ }
+
@Override
public OutputStream create(StoragePath path) throws IOException {
return fs.create(convertToHadoopPath(path));
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
index 7eb2901c1d3..89bb52a0765 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
@@ -22,8 +22,8 @@ import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HoodieRetryWrapperFileSystem;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
@@ -77,7 +77,7 @@ public class TestFSUtilsWithRetryWrapperEnable extends TestFSUtils {
HoodieWrapperFileSystem fs =
new HoodieWrapperFileSystem(fileSystem, new NoOpConsistencyGuard());
- HoodieStorage storage = HoodieStorageUtils.getStorage(fs);
+ HoodieStorage storage = new HoodieHadoopStorage(fs);
metaClient.setHoodieStorage(storage);
}
@@ -91,7 +91,7 @@ public class TestFSUtilsWithRetryWrapperEnable extends TestFSUtils {
initialRetryIntervalMs, "");
HoodieWrapperFileSystem fs =
new HoodieWrapperFileSystem(fileSystem, new NoOpConsistencyGuard());
- HoodieStorage storage = HoodieStorageUtils.getStorage(fs);
+ HoodieStorage storage = new HoodieHadoopStorage(fs);
metaClient.setHoodieStorage(storage);
List<String> folders =
Arrays.asList("2016/04/15", ".hoodie/.temp/2/2016/04/15");
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestHoodieWrapperFileSystem.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestHoodieWrapperFileSystem.java
index c7b5217524e..587989216d6 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestHoodieWrapperFileSystem.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestHoodieWrapperFileSystem.java
@@ -24,9 +24,9 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -77,7 +77,7 @@ class TestHoodieWrapperFileSystem {
StoragePath testFile = new StoragePath(basePath + StoragePath.SEPARATOR + "clean.00000001");
// create same commit twice
- HoodieStorage storage = HoodieStorageUtils.getStorage(fs);
+ HoodieStorage storage = new HoodieHadoopStorage(fs);
storage.createImmutableFileInPath(testFile, Option.of(getUTF8Bytes(testContent)));
storage.createImmutableFileInPath(testFile, Option.of(getUTF8Bytes(testContent)));
List<StoragePathInfo> pathInfoList = storage.listDirectEntries(new StoragePath(basePath));
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
index ef699cd4937..7b884ca70cf 100755
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
@@ -65,6 +65,7 @@ import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
@@ -145,11 +146,11 @@ public class TestHoodieLogFormat extends HoodieCommonTestHarness {
@BeforeAll
public static void setUpClass() throws IOException {
if (shouldUseExternalHdfs()) {
- storage = HoodieStorageUtils.getStorage(useExternalHdfs());
+ storage = new HoodieHadoopStorage(useExternalHdfs());
} else {
// Append is not supported in LocalFileSystem. HDFS needs to be setup.
hdfsTestService = new HdfsTestService();
- storage = HoodieStorageUtils.getStorage(hdfsTestService.start(true).getFileSystem());
+ storage = new HoodieHadoopStorage(hdfsTestService.start(true).getFileSystem());
}
}
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormatAppendFailure.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormatAppendFailure.java
index 038bcf93cf5..c50c46485c3 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormatAppendFailure.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormatAppendFailure.java
@@ -28,8 +28,8 @@ import org.apache.hudi.common.table.log.block.HoodieCommandBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.testutils.SchemaTestUtil;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -103,7 +103,7 @@ public class TestHoodieLogFormatAppendFailure {
// Use some fs like LocalFileSystem, that does not support appends
String uuid = UUID.randomUUID().toString();
StoragePath localPartitionPath = new StoragePath("/tmp/");
- HoodieStorage storage = HoodieStorageUtils.getStorage(cluster.getFileSystem());
+ HoodieStorage storage = new HoodieHadoopStorage(cluster.getFileSystem());
StoragePath testPath = new StoragePath(localPartitionPath, uuid);
storage.createDirectory(testPath);
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
index a317d616136..fa2d7558ef5 100755
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
@@ -29,8 +29,8 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hadoop.fs.FileSystem;
import org.junit.jupiter.api.AfterEach;
@@ -709,7 +709,7 @@ public class TestHoodieActiveTimeline extends HoodieCommonTestHarness {
HoodieStorage storage = metaClient.getStorage();
FileSystem fs = (FileSystem) storage.getFileSystem();
HoodieWrapperFileSystem newFs = new HoodieWrapperFileSystem(fs, new NoOpConsistencyGuard());
- metaClient.setHoodieStorage(HoodieStorageUtils.getStorage(newFs));
+ metaClient.setHoodieStorage(new HoodieHadoopStorage(newFs));
try {
fun.accept(metaClient);
} finally {
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieAvroFileReaderFactory.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieAvroFileReaderFactory.java
index 85731674cd6..b46e0747aeb 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieAvroFileReaderFactory.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieAvroFileReaderFactory.java
@@ -50,7 +50,7 @@ public class TestHoodieAvroFileReaderFactory {
final StorageConfiguration<?> storageConf = HadoopFSUtils.getStorageConf(new Configuration());
final StoragePath parquetPath = new StoragePath("/partition/path/f1_1-0-1_000.parquet");
HoodieFileReader parquetReader = HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, storageConf, parquetPath);
+ .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, parquetPath);
assertTrue(parquetReader instanceof HoodieAvroParquetReader);
// log file format.
@@ -58,7 +58,7 @@ public class TestHoodieAvroFileReaderFactory {
"/partition/path/f.b51192a8-574b-4a85-b246-bcfec03ac8bf_100.log.2_1-0-1");
final Throwable thrown = assertThrows(UnsupportedOperationException.class, () -> {
HoodieFileReader logWriter = HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, storageConf, logPath);
+ .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, logPath);
}, "should fail since log storage reader is not supported yet.");
assertTrue(thrown.getMessage().contains("format not supported yet."));
@@ -66,7 +66,7 @@ public class TestHoodieAvroFileReaderFactory {
final StoragePath orcPath = new StoragePath("/partition/path/f1_1-0-1_000.orc");
HoodieFileReader orcReader = HoodieIOFactory.getIOFactory(storageConf)
.getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, storageConf, orcPath);
+ .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, orcPath);
assertTrue(orcReader instanceof HoodieAvroOrcReader);
}
}
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieOrcReaderWriter.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieOrcReaderWriter.java
index 1fec959ba93..314334365b2 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieOrcReaderWriter.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieOrcReaderWriter.java
@@ -79,7 +79,7 @@ public class TestHoodieOrcReaderWriter extends TestHoodieReaderWriterBase {
protected HoodieAvroFileReader createReader(
StorageConfiguration<?> conf) throws Exception {
return (HoodieAvroFileReader) HoodieIOFactory.getIOFactory(conf).getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, conf, getFilePath());
+ .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, getFilePath());
}
@Override
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieHFileRecordReader.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieHFileRecordReader.java
index 85e9fcac311..1d057901908 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieHFileRecordReader.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieHFileRecordReader.java
@@ -60,7 +60,7 @@ public class HoodieHFileRecordReader implements RecordReader<NullWritable, Array
StorageConfiguration<?> storageConf = HadoopFSUtils.getStorageConf(conf);
HoodieConfig hoodieConfig = getReaderConfigs(storageConf);
reader = HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(HoodieRecord.HoodieRecordType.AVRO)
- .getFileReader(hoodieConfig, HadoopFSUtils.getStorageConf(conf), path, HoodieFileFormat.HFILE, Option.empty());
+ .getFileReader(hoodieConfig, path, HoodieFileFormat.HFILE, Option.empty());
schema = reader.getSchema();
valueObj = new ArrayWritable(Writable.class, new Writable[schema.getFields().size()]);
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java
index 79829cc3917..8884e0a3c06 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java
@@ -37,8 +37,8 @@ import org.apache.hudi.internal.schema.action.InternalSchemaMerger;
import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
import org.apache.hudi.internal.schema.utils.InternalSchemaUtils;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.avro.Schema;
import org.apache.hadoop.fs.FileSystem;
@@ -117,7 +117,7 @@ public class SchemaEvolutionContext {
try {
Path inputPath = ((FileSplit) split).getPath();
FileSystem fs = inputPath.getFileSystem(job);
- HoodieStorage storage = HoodieStorageUtils.getStorage(fs);
+ HoodieStorage storage = new HoodieHadoopStorage(fs);
Option<StoragePath> tablePath = TablePathUtils.getTablePath(storage, convertToStoragePath(inputPath));
return HoodieTableMetaClient.builder().setBasePath(tablePath.get().toString())
.setConf(HadoopFSUtils.getStorageConfWithCopy(job)).build();
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
index 6d4b79c6896..a612ab4616c 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
@@ -310,7 +310,7 @@ public class HoodieRealtimeRecordReaderUtils {
StorageConfiguration<?> storageConf = HadoopFSUtils.getStorageConf(conf);
HoodieConfig hoodieConfig = getReaderConfigs(storageConf);
return HoodieIOFactory.getIOFactory(storageConf).getReaderFactory(HoodieRecord.HoodieRecordType.AVRO)
- .getFileReader(hoodieConfig, HadoopFSUtils.getStorageConf(conf), convertToStoragePath(path));
+ .getFileReader(hoodieConfig, convertToStoragePath(path));
}
private static Schema appendNullSchemaFields(Schema schema, List<String> newFieldNames) {
diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java
index c19bd7f5a1e..3371b5efb27 100644
--- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java
+++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java
@@ -34,8 +34,8 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat;
import org.apache.hudi.hadoop.testutils.InputFormatTestUtil;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
@@ -86,7 +86,7 @@ public class TestHoodieCombineHiveInputFormat extends HoodieCommonTestHarness {
// Append is not supported in LocalFileSystem. HDFS needs to be setup.
hdfsTestService = new HdfsTestService();
fs = hdfsTestService.start(true).getFileSystem();
- storage = HoodieStorageUtils.getStorage(fs);
+ storage = new HoodieHadoopStorage(fs);
}
@AfterAll
diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadSnapshotReader.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadSnapshotReader.java
index 15a935bbd9e..7ba8e78ceed 100644
--- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadSnapshotReader.java
+++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadSnapshotReader.java
@@ -37,10 +37,11 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.hadoop.config.HoodieRealtimeConfig;
+import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.testutils.InputFormatTestUtil;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
@@ -88,7 +89,7 @@ public class TestHoodieMergeOnReadSnapshotReader {
baseJobConf.set(HoodieRealtimeConfig.MAX_DFS_STREAM_BUFFER_SIZE_PROP, String.valueOf(1024 * 1024));
baseJobConf.set(serdeConstants.LIST_COLUMNS, COLUMNS);
baseJobConf.set(serdeConstants.LIST_COLUMN_TYPES, COLUMN_TYPES);
- storage = HoodieStorageUtils.getStorage(basePath.toUri().toString(), baseJobConf);
+ storage = new HoodieHadoopStorage(HadoopFSUtils.getFs(new StoragePath(basePath.toUri()), baseJobConf));
}
@AfterEach
@@ -112,7 +113,7 @@ public class TestHoodieMergeOnReadSnapshotReader {
private void testReaderInternal(boolean partitioned, HoodieLogBlock.HoodieLogBlockType logBlockType) throws Exception {
// initial commit
Schema schema = HoodieAvroUtils.addMetadataFields(SchemaTestUtil.getEvolvedSchema());
- HoodieTestUtils.init(HoodieStorageUtils.getStorageConf(hadoopConf), basePath.toString(), HoodieTableType.MERGE_ON_READ);
+ HoodieTestUtils.init(HadoopFSUtils.getStorageConf(hadoopConf), basePath.toString(), HoodieTableType.MERGE_ON_READ);
String baseInstant = "100";
File partitionDir = partitioned ? InputFormatTestUtil.prepareParquetTable(basePath, schema, 1, TOTAL_RECORDS, baseInstant,
HoodieTableType.MERGE_ON_READ)
diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadTableInputFormat.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadTableInputFormat.java
index 05ab9787614..8824adc1e34 100644
--- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadTableInputFormat.java
+++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadTableInputFormat.java
@@ -23,8 +23,8 @@ import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.hadoop.PathWithBootstrapFileStatus;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -52,7 +52,7 @@ public class TestHoodieMergeOnReadTableInputFormat {
@BeforeEach
void setUp() throws IOException {
fs = FileSystem.get(tempDir.toUri(), new Configuration());
- storage = HoodieStorageUtils.getStorage(fs);
+ storage = new HoodieHadoopStorage(fs);
}
@AfterEach
diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieRealtimeRecordReader.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieRealtimeRecordReader.java
index c05e6e9d128..adc6c5b83fc 100644
--- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieRealtimeRecordReader.java
+++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieRealtimeRecordReader.java
@@ -47,8 +47,8 @@ import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.testutils.InputFormatTestUtil;
import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
@@ -120,7 +120,7 @@ public class TestHoodieRealtimeRecordReader {
baseJobConf = new JobConf(storageConf.unwrap());
baseJobConf.set(HoodieRealtimeConfig.MAX_DFS_STREAM_BUFFER_SIZE_PROP, String.valueOf(1024 * 1024));
fs = HadoopFSUtils.getFs(basePath.toUri().toString(), baseJobConf);
- storage = HoodieStorageUtils.getStorage(fs);
+ storage = new HoodieHadoopStorage(fs);
}
@AfterEach
diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java
index 540932003d7..7cdf3e6af29 100644
--- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java
+++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java
@@ -39,8 +39,8 @@ import org.apache.hudi.common.testutils.SchemaTestUtil;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.hadoop.utils.HoodieHiveUtils;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
@@ -506,7 +506,7 @@ public class InputFormatTestUtil {
HoodiePartitionMetadata partitionMetadata =
new HoodiePartitionMetadata(
- HoodieStorageUtils.getStorage(new LocalFileSystem(lfs)),
+ new HoodieHadoopStorage(new LocalFileSystem(lfs)),
"0",
new StoragePath(basePath.toAbsolutePath().toString()),
new StoragePath(partitionPath.toAbsolutePath().toString()),
diff --git a/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/reader/DFSHoodieDatasetInputReader.java b/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/reader/DFSHoodieDatasetInputReader.java
index fd3cc287323..6c3286a47bf 100644
--- a/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/reader/DFSHoodieDatasetInputReader.java
+++ b/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/reader/DFSHoodieDatasetInputReader.java
@@ -278,7 +278,6 @@ public class DFSHoodieDatasetInputReader extends DFSDeltaInputReader {
.getReaderFactory(HoodieRecordType.AVRO)
.getFileReader(
DEFAULT_HUDI_CONFIG_FOR_READER,
- metaClient.getStorageConf(),
fileSlice.getBaseFile().get().getStoragePath()));
return new CloseableMappingIterator<>(reader.getRecordIterator(schema), HoodieRecord::getData);
} else {
diff --git a/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java b/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java
index 586b5b0a56f..fcc8d2d505d 100644
--- a/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java
+++ b/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java
@@ -279,6 +279,12 @@ public abstract class HoodieStorage implements Closeable {
@PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
public abstract Object unwrapConf();
+ /**
+ * @return the raw storage.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ public abstract HoodieStorage getRawStorage();
+
/**
* Creates a new file with overwrite set to false. This ensures files are created
* only once and never rewritten, also, here we take care if the content is not
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
index a6f661c9e46..fc03a26ac82 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
@@ -758,8 +758,8 @@ object HoodieBaseRelation extends SparkAdapterSupport {
val hoodieConfig = new HoodieConfig()
hoodieConfig.setValue(USE_NATIVE_HFILE_READER,
options.getOrElse(USE_NATIVE_HFILE_READER.key(), USE_NATIVE_HFILE_READER.defaultValue().toString))
- val reader = (new HoodieSparkIOFactory).getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(hoodieConfig, storageConf, filePath, HFILE)
+ val reader = new HoodieSparkIOFactory(storageConf).getReaderFactory(HoodieRecordType.AVRO)
+ .getFileReader(hoodieConfig, filePath, HFILE)
val requiredRowSchema = requiredDataSchema.structTypeSchema
// NOTE: Schema has to be parsed at this point, since Avro's [[Schema]] aren't serializable
diff --git a/hudi-spark-datasource/hudi-spark/src/test/java/HoodieJavaStreamingApp.java b/hudi-spark-datasource/hudi-spark/src/test/java/HoodieJavaStreamingApp.java
index d0e1b44e439..086363e447c 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/java/HoodieJavaStreamingApp.java
+++ b/hudi-spark-datasource/hudi-spark/src/test/java/HoodieJavaStreamingApp.java
@@ -30,6 +30,7 @@ import org.apache.hudi.exception.TableNotFoundException;
import org.apache.hudi.hive.HiveSyncConfig;
import org.apache.hudi.hive.MultiPartKeysValueExtractor;
import org.apache.hudi.hive.SlashEncodedDayPartitionValueExtractor;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.testutils.HoodieClientTestUtils;
import com.beust.jcommander.JCommander;
@@ -263,7 +264,7 @@ public class HoodieJavaStreamingApp {
if (timeline.countInstants() >= numCommits) {
return;
}
- HoodieTableMetaClient metaClient = createMetaClient(fs.getConf(), tablePath);
+ HoodieTableMetaClient metaClient = createMetaClient(new HadoopStorageConfiguration(fs.getConf()), tablePath);
System.out.println("Instants :" + metaClient.getActiveTimeline().getInstants());
} catch (TableNotFoundException te) {
LOG.info("Got table not found exception. Retrying");
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSourceStorage.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSourceStorage.scala
index 79de5800217..ad017a5a4dc 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSourceStorage.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSourceStorage.scala
@@ -26,10 +26,10 @@ import org.apache.hudi.common.testutils.{HoodieTestDataGenerator, HoodieTestUtil
import org.apache.hudi.common.util.StringUtils
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.hadoop.fs.HadoopFSUtils
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness.getSparkSqlConf
import org.apache.hudi.{DataSourceReadOptions, DataSourceWriteOptions, HoodieDataSourceHelpers}
-
import org.apache.spark.SparkConf
import org.apache.spark.sql._
import org.apache.spark.sql.functions.{col, lit}
@@ -176,7 +176,7 @@ class TestMORDataSourceStorage extends SparkClientFunctionalTestHarness {
.save(basePath)
}
// compaction should have been completed
- val metaClient = HoodieTestUtils.createMetaClient(fs.getConf, basePath)
+ val metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(fs.getConf), basePath)
assertEquals(1, metaClient.getActiveTimeline.getCommitTimeline.countInstants())
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestClusteringProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestClusteringProcedure.scala
index e60a08fa197..ee05cbcaf3c 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestClusteringProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestClusteringProcedure.scala
@@ -27,9 +27,9 @@ import org.apache.hudi.common.testutils.HoodieTestUtils
import org.apache.hudi.common.util.collection.Pair
import org.apache.hudi.common.util.{Option => HOption}
import org.apache.hudi.{DataSourceReadOptions, HoodieCLIUtils, HoodieDataSourceHelpers, HoodieFileIndex}
-
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, Literal}
import org.apache.spark.sql.types.{DataTypes, Metadata, StringType, StructField, StructType}
import org.apache.spark.sql.{Dataset, Row}
@@ -441,7 +441,7 @@ class TestClusteringProcedure extends HoodieSparkProcedureTestBase {
spark.sql(s"call run_clustering(table => '$tableName', op => 'schedule')")
val conf = new Configuration
- val metaClient = HoodieTestUtils.createMetaClient(conf, basePath)
+ val metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(conf), basePath)
val instants = metaClient.getActiveTimeline.filterPendingReplaceTimeline().getInstants.iterator().asScala.map(_.getTimestamp).toSeq
assert(2 == instants.size)
@@ -505,7 +505,7 @@ class TestClusteringProcedure extends HoodieSparkProcedureTestBase {
writeRecords(2, 4, 0, basePath, Map("hoodie.avro.schema.validate"-> "false"))
val conf = new Configuration
- val metaClient = HoodieTestUtils.createMetaClient(conf, basePath)
+ val metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(conf), basePath)
assert(0 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
assert(metaClient.getActiveTimeline.filterPendingReplaceTimeline().empty())
@@ -576,7 +576,7 @@ class TestClusteringProcedure extends HoodieSparkProcedureTestBase {
// insert records
writeRecords(fileNum, numRecords, 0, basePath, metadataOpts ++ Map("hoodie.avro.schema.validate"-> "false"))
val conf = new Configuration
- val metaClient = HoodieTestUtils.createMetaClient(conf, basePath)
+ val metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(conf), basePath)
val avgSize = avgRecord(metaClient.getActiveTimeline)
val avgCount = Math.ceil(1.0 * numRecords / fileNum).toLong
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala
index 606fc8566a9..1465ceefe20 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala
@@ -22,8 +22,8 @@ package org.apache.spark.sql.hudi.procedure
import org.apache.hudi.common.table.timeline.HoodieInstant
import org.apache.hudi.common.testutils.HoodieTestUtils
import org.apache.hudi.common.testutils.HoodieTestUtils.createMetaClient
-
import org.apache.hadoop.conf.Configuration
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
class TestCompactionProcedure extends HoodieSparkProcedureTestBase {
diff --git a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
index 136c9c4e636..af4baaae4a3 100644
--- a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
+++ b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
@@ -42,6 +42,7 @@ import org.apache.hudi.hive.ddl.HMSDDLExecutor;
import org.apache.hudi.hive.ddl.HiveSyncMode;
import org.apache.hudi.hive.testutils.HiveTestUtil;
import org.apache.hudi.hive.util.IMetaStoreClientUtil;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.sync.common.HoodieSyncConfig;
import org.apache.hudi.sync.common.model.FieldSchema;
import org.apache.hudi.sync.common.model.Partition;
@@ -450,7 +451,7 @@ public class TestHiveSyncTool {
HiveTestUtil.removeCommitFromActiveTimeline("500", COMMIT_ACTION);
HiveTestUtil.removeCommitFromActiveTimeline("600", COMMIT_ACTION);
HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(
- hiveClient.config.getHadoopConf(), basePath);
+ new HadoopStorageConfiguration(hiveClient.config.getHadoopConf()), basePath);
assertEquals(
Arrays.asList("400", "700", "800"),
metaClient.getActiveTimeline().getInstants().stream()
diff --git a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java
index f5eab7f87e5..0d55ac09309 100644
--- a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java
+++ b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java
@@ -56,8 +56,8 @@ import org.apache.hudi.hive.ddl.HiveQueryDDLExecutor;
import org.apache.hudi.hive.ddl.QueryBasedDDLExecutor;
import org.apache.hudi.hive.util.IMetaStoreClientUtil;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.avro.Schema;
import org.apache.avro.generic.IndexedRecord;
@@ -162,7 +162,7 @@ public class HiveTestUtil {
hiveSyncConfig = new HiveSyncConfig(hiveSyncProps, hiveTestService.getHiveConf());
fileSystem = hiveSyncConfig.getHadoopFileSystem();
- storage = HoodieStorageUtils.getStorage(fileSystem);
+ storage = new HoodieHadoopStorage(fileSystem);
dtfOut = DateTimeFormatter.ofPattern("yyyy/MM/dd");
if (ddlExecutor != null) {
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
index 7ceaddeeb12..c2237e32cee 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
@@ -1488,8 +1488,9 @@ public class HoodieMetadataTableValidator implements Serializable {
HoodieConfig hoodieConfig = new HoodieConfig();
hoodieConfig.setValue(HoodieReaderConfig.USE_NATIVE_HFILE_READER,
Boolean.toString(ConfigUtils.getBooleanWithAltKeys(props, HoodieReaderConfig.USE_NATIVE_HFILE_READER)));
- try (HoodieFileReader fileReader = getHoodieSparkIOFactory().getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(hoodieConfig, metaClient.getStorageConf(), path)) {
+ try (HoodieFileReader fileReader = getHoodieSparkIOFactory(metaClient.getStorageConf())
+ .getReaderFactory(HoodieRecordType.AVRO)
+ .getFileReader(hoodieConfig, path)) {
bloomFilter = fileReader.readBloomFilter();
if (bloomFilter == null) {
LOG.error("Failed to read bloom filter for {}", path);
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
index 5c29a981252..c8a1b47b9fb 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
@@ -23,7 +23,7 @@ import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.Option;
-import org.apache.hudi.storage.HoodieStorageUtils;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.streamer.DefaultStreamContext;
import org.apache.hudi.utilities.streamer.HoodieStreamer;
@@ -53,6 +53,6 @@ public class DeltaSync extends StreamSync {
TypedProperties props, HoodieSparkEngineContext hoodieSparkContext, FileSystem fs, Configuration conf,
Function<SparkRDDWriteClient, Boolean> onInitializingHoodieWriteClient) throws IOException {
super(cfg, sparkSession, props, hoodieSparkContext,
- HoodieStorageUtils.getStorage(fs), conf, onInitializingHoodieWriteClient, new DefaultStreamContext(schemaProvider, Option.empty()));
+ new HoodieHadoopStorage(fs), conf, onInitializingHoodieWriteClient, new DefaultStreamContext(schemaProvider, Option.empty()));
}
}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java
index 34288b0a0d3..6c5cca9888e 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java
@@ -21,7 +21,7 @@ package org.apache.hudi.utilities.deltastreamer;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.Option;
-import org.apache.hudi.storage.HoodieStorageUtils;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.utilities.streamer.HoodieStreamer;
import org.apache.hadoop.conf.Configuration;
@@ -51,7 +51,7 @@ public class HoodieDeltaStreamer extends HoodieStreamer {
JavaSparkContext jssc,
FileSystem fs,
Configuration conf) throws IOException {
- super(cfg, jssc, HoodieStorageUtils.getStorage(fs), conf);
+ super(cfg, jssc, new HoodieHadoopStorage(fs), conf);
}
public HoodieDeltaStreamer(Config cfg,
@@ -59,7 +59,7 @@ public class HoodieDeltaStreamer extends HoodieStreamer {
FileSystem fs,
Configuration conf,
Option<TypedProperties> propsOverride) throws IOException {
- super(cfg, jssc, HoodieStorageUtils.getStorage(fs), conf, propsOverride);
+ super(cfg, jssc, new HoodieHadoopStorage(fs), conf, propsOverride);
}
@Deprecated
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
index 87712243bd7..3bc937836f2 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
@@ -75,9 +75,9 @@ import org.apache.hudi.keygen.KeyGenUtils;
import org.apache.hudi.keygen.factory.HoodieSparkKeyGeneratorFactory;
import org.apache.hudi.metrics.HoodieMetrics;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.sync.common.util.SyncUtilHelpers;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.apache.hudi.util.JavaScalaConverters;
@@ -291,7 +291,7 @@ public class StreamSync implements Serializable, Closeable {
TypedProperties props, JavaSparkContext jssc, FileSystem fs, Configuration conf,
Function<SparkRDDWriteClient, Boolean> onInitializingHoodieWriteClient) throws IOException {
this(cfg, sparkSession, props, new HoodieSparkEngineContext(jssc),
- HoodieStorageUtils.getStorage(fs), conf, onInitializingHoodieWriteClient,
+ new HoodieHadoopStorage(fs), conf, onInitializingHoodieWriteClient,
new DefaultStreamContext(schemaProvider, Option.empty()));
}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerWithMultiWriter.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerWithMultiWriter.java
index 04998bc7e99..5060bb2545a 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerWithMultiWriter.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerWithMultiWriter.java
@@ -32,6 +32,7 @@ import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.execution.bulkinsert.BulkInsertSortMode;
import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.utilities.config.SourceTestConfig;
import org.apache.hudi.utilities.sources.TestDataSource;
import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
@@ -142,7 +143,7 @@ public class TestHoodieDeltaStreamerWithMultiWriter extends HoodieDeltaStreamerT
HoodieDeltaStreamer.Config cfgBackfillJob = getDeltaStreamerConfig(tableBasePath, tableType.name(), WriteOperationType.UPSERT,
propsFilePath, Collections.singletonList(TestHoodieDeltaStreamer.TripsWithDistanceTransformer.class.getName()));
cfgBackfillJob.continuousMode = false;
- HoodieTableMetaClient meta = createMetaClient(hadoopConf, tableBasePath);
+ HoodieTableMetaClient meta = createMetaClient(new HadoopStorageConfiguration(hadoopConf), tableBasePath);
HoodieTimeline timeline = meta.reloadActiveTimeline().getCommitsTimeline().filterCompletedInstants();
HoodieCommitMetadata commitMetadata = HoodieCommitMetadata
.fromBytes(timeline.getInstantDetails(timeline.firstInstant().get()).get(), HoodieCommitMetadata.class);
@@ -210,7 +211,7 @@ public class TestHoodieDeltaStreamerWithMultiWriter extends HoodieDeltaStreamerT
HoodieDeltaStreamer.Config cfgBackfillJob2 = getDeltaStreamerConfig(tableBasePath, tableType.name(), WriteOperationType.INSERT,
propsFilePath, Collections.singletonList(TestHoodieDeltaStreamer.TestIdentityTransformer.class.getName()));
cfgBackfillJob2.continuousMode = false;
- HoodieTableMetaClient meta = createMetaClient(hadoopConf, tableBasePath);
+ HoodieTableMetaClient meta = createMetaClient(new HadoopStorageConfiguration(hadoopConf), tableBasePath);
HoodieTimeline timeline = meta.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
HoodieCommitMetadata commitMetadata = HoodieCommitMetadata
.fromBytes(timeline.getInstantDetails(timeline.firstInstant().get()).get(), HoodieCommitMetadata.class);
@@ -389,7 +390,7 @@ public class TestHoodieDeltaStreamerWithMultiWriter extends HoodieDeltaStreamerT
HoodieDeltaStreamer ingestionJob, HoodieDeltaStreamer.Config cfgIngestionJob, HoodieDeltaStreamer backfillJob,
HoodieDeltaStreamer.Config cfgBackfillJob, boolean expectConflict, String jobId) throws Exception {
ExecutorService service = Executors.newFixedThreadPool(2);
- HoodieTableMetaClient meta = createMetaClient(hadoopConf, tableBasePath);
+ HoodieTableMetaClient meta = createMetaClient(new HadoopStorageConfiguration(hadoopConf), tableBasePath);
HoodieTimeline timeline = meta.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
String lastSuccessfulCommit = timeline.lastInstant().get().getTimestamp();
// Condition for parallel ingestion job
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestStreamSyncUnitTests.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestStreamSyncUnitTests.java
index fe775f95a36..f429943532f 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestStreamSyncUnitTests.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestStreamSyncUnitTests.java
@@ -29,7 +29,7 @@ import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieErrorTableConfig;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.InputBatch;
import org.apache.hudi.utilities.transform.Transformer;
@@ -71,7 +71,7 @@ public class TestStreamSyncUnitTests {
Boolean isNullTargetSchema, Boolean hasErrorTable, Boolean shouldTryWriteToErrorTable) {
//basic deltastreamer inputs
HoodieSparkEngineContext hoodieSparkEngineContext = mock(HoodieSparkEngineContext.class);
- HoodieStorage storage = HoodieStorageUtils.getStorage(mock(FileSystem.class));
+ HoodieStorage storage = new HoodieHadoopStorage(mock(FileSystem.class));
SparkSession sparkSession = mock(SparkSession.class);
Configuration configuration = mock(Configuration.class);
HoodieStreamer.Config cfg = new HoodieStreamer.Config();
@@ -141,7 +141,7 @@ public class TestStreamSyncUnitTests {
@MethodSource("getCheckpointToResumeCases")
void testGetCheckpointToResume(HoodieStreamer.Config cfg, HoodieCommitMetadata commitMetadata, Option<String> expectedResumeCheckpoint) throws IOException {
HoodieSparkEngineContext hoodieSparkEngineContext = mock(HoodieSparkEngineContext.class);
- HoodieStorage storage = HoodieStorageUtils.getStorage(mock(FileSystem.class));
+ HoodieStorage storage = new HoodieHadoopStorage(mock(FileSystem.class));
TypedProperties props = new TypedProperties();
SparkSession sparkSession = mock(SparkSession.class);
Configuration configuration = mock(Configuration.class);
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/UtilitiesTestBase.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/UtilitiesTestBase.java
index 762238c4674..ba9746302fb 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/UtilitiesTestBase.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/UtilitiesTestBase.java
@@ -38,8 +38,8 @@ import org.apache.hudi.hive.ddl.JDBCExecutor;
import org.apache.hudi.hive.ddl.QueryBasedDDLExecutor;
import org.apache.hudi.hive.testutils.HiveTestService;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.utilities.UtilHelpers;
import org.apache.hudi.utilities.sources.TestDataSource;
@@ -152,7 +152,7 @@ public class UtilitiesTestBase {
fs = FileSystem.getLocal(hadoopConf);
basePath = sharedTempDir.toUri().toString();
}
- storage = HoodieStorageUtils.getStorage(fs);
+ storage = new HoodieHadoopStorage(fs);
hadoopConf.set("hive.exec.scratchdir", basePath + "/.tmp/hive");
if (needsHive) {