This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new c28a87d8987c fix(storage): resolve HoodieStorage from a path instead
of a default local URI (#19737)
c28a87d8987c is described below
commit c28a87d8987ca7f8b27904438430c5158c864b62
Author: Y Ethan Guo <[email protected]>
AuthorDate: Wed Aug 26 09:06:23 2026 -0700
fix(storage): resolve HoodieStorage from a path instead of a default local
URI (#19737)
---
.../hudi/cli/commands/TestCommitsCommand.java | 3 +-
.../table/action/rollback/RollbackHelperV1.java | 3 -
.../action/rollback/TestRollbackHelperV1.java | 63 +--------
.../upgrade/TestEightToSevenDowngradeHandler.java | 5 +-
.../table/functional/TestCleanActionExecutor.java | 3 +-
.../hudi/common/util/HoodieStorageUtils.java | 12 +-
.../hudi/common/testutils/HoodieTestUtils.java | 16 ++-
.../hadoop/fs/NonLocalSchemeLocalFileSystem.java | 88 ++++++++++++
.../hudi/io/hadoop/TestHoodieHadoopIOFactory.java | 2 +-
.../spark/sql/hudi/blob/BatchedBlobReader.scala | 45 +++++--
.../TestIncrementalQueryWithArchivedInstants.scala | 4 +-
.../org/apache/hudi/blob/BlobTestHelpers.scala | 35 +++++
.../apache/hudi/blob/CountingHoodieStorage.scala | 55 ++++++++
.../apache/hudi/blob/TestBatchedBlobReader.scala | 150 +++++++++++++++++++++
.../org/apache/hudi/blob/TestReadBlobSQL.scala | 55 ++++++++
.../sql/hudi/blob/TestBatchedBlobReaderMerge.scala | 4 +-
.../org/apache/hudi/utilities/TestUtilHelpers.java | 7 +-
.../utilities/sources/TestHoodieIncrSource.java | 5 +-
.../streamer/TestHoodieStreamerMetrics.java | 17 ++-
19 files changed, 467 insertions(+), 105 deletions(-)
diff --git
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java
index 8db4ab321c7b..54f2de8bc835 100644
---
a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java
+++
b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java
@@ -42,7 +42,6 @@ import
org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.testutils.InProcessTimeGenerator;
-import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.NumericUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
@@ -170,7 +169,7 @@ public class TestCommitsCommand extends
CLIFunctionalTestHarness {
}
private String generateExpectData(int records, Map<String, Integer[]> data)
throws IOException {
- HoodieStorage storage = HoodieStorageUtils.getStorage(storageConf());
+ HoodieStorage storage = HoodieTestUtils.getLocalStorage(storageConf());
List<String> partitionPaths =
FSUtils.getAllPartitionFoldersThreeLevelsDown(storage, tablePath1);
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java
index 9b47d86e1240..da8170006f6b 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java
@@ -502,9 +502,6 @@ public class RollbackHelperV1 extends RollbackHelper {
// fetch file sizes.
StoragePath fullPartitionPath =
StringUtils.isNullOrEmpty(partition) ? new StoragePath(basePathStr) : new
StoragePath(basePathStr, partition);
- // Resolve storage from the partition path, not the no-path
overload: the latter binds to
- // HoodieStorageUtils.DEFAULT_URI ("file:///"), so listing an
s3a/gs partition on an executor
- // throws "Wrong FS ... expected: file:///" and the rollback can
never complete.
HoodieStorage storage =
HoodieStorageUtils.getStorage(fullPartitionPath, storageConfiguration);
List<Option<StoragePathInfo>> storagePathInfoOpts =
getPathInfoUnderPartition(storage,
fullPartitionPath, new HashSet<>(missingLogFiles), true);
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackHelperV1.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackHelperV1.java
index 05008aef58ef..2ebd21b3e086 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackHelperV1.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackHelperV1.java
@@ -32,6 +32,7 @@ import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.hadoop.fs.NonLocalSchemeLocalFileSystem;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
@@ -39,8 +40,6 @@ import org.apache.hudi.storage.StoragePathInfo;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.RawLocalFileSystem;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -49,7 +48,6 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.io.OutputStream;
-import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashSet;
@@ -69,7 +67,7 @@ import static org.mockito.Mockito.when;
* Covers {@link RollbackHelperV1} on a table whose base path is NOT on the
local filesystem.
*
* <p>Every other rollback test runs the table under a local {@code file://}
base path, where the
- * default storage URI of {@link HoodieStorageUtils#DEFAULT_URI} happens to be
the correct
+ * default storage URI of {@link HoodieTestUtils#DEFAULT_URI} happens to be
the correct
* filesystem. That is precisely why a filesystem-resolution defect could
survive unnoticed in this
* class. These tests put the table under an object-store scheme instead,
which is how the defect
* shows up in production as {@code IllegalArgumentException: Wrong FS:
s3a://..., expected:
@@ -88,55 +86,6 @@ class TestRollbackHelperV1 extends HoodieRollbackTestBase {
private static final byte[] LOG_FILE_CONTENT =
"not a real log block, only bytes to
size".getBytes(StandardCharsets.UTF_8);
- /**
- * The local filesystem exposed under a scheme other than {@code file}, so a
test can tell apart
- * storage resolved from a path (correct) and storage resolved from the
default URI (wrong), the
- * way s3a and gs do in production without needing a remote object store.
- *
- * <p>{@link RawLocalFileSystem#pathToFile} keeps only the path component of
a URI, so a
- * {@code <scheme>://<bucket>/tmp/x} path reads and writes the local file
{@code /tmp/x}. What the
- * subclass changes is only the identity the filesystem reports, which is
what
- * {@link FileSystem#checkPath} validates every path against.
- */
- public static class NonLocalSchemeLocalFileSystem extends RawLocalFileSystem
{
- /**
- * Answer for {@link #getUri()} until {@link #initialize} supplies the
real one. The superclass
- * constructor calls {@code getUri()}, which runs before any instance
field of this subclass is
- * assigned, so the fallback has to be a static.
- */
- private static final URI UNINITIALIZED_URI = URI.create(SCHEME + ":///");
-
- private URI uri;
-
- @Override
- public void initialize(URI name, Configuration conf) throws IOException {
- super.initialize(name, conf);
- this.uri = URI.create(name.getScheme() + "://"
- + (name.getAuthority() == null ? "" : name.getAuthority()));
- setWorkingDirectory(new Path(this.uri.toString() + Path.SEPARATOR));
- }
-
- @Override
- public URI getUri() {
- return uri == null ? UNINITIALIZED_URI : uri;
- }
-
- @Override
- public String getScheme() {
- return getUri().getScheme();
- }
-
- /**
- * The superclass qualifies the process working directory against {@link
#getUri()} from its own
- * constructor. Skip that: this filesystem does not know its real URI yet,
and every path these
- * tests use is absolute, so the working directory is never consulted.
- */
- @Override
- protected Path getInitialWorkingDirectory() {
- return new Path(System.getProperty("user.dir"));
- }
- }
-
@Override
protected StoragePath createBasePath() {
return new StoragePath(SCHEME + "://" + BUCKET + tmpDir + "/" +
UUID.randomUUID());
@@ -181,9 +130,9 @@ class TestRollbackHelperV1 extends HoodieRollbackTestBase {
assertEquals(LOG_FILE_CONTENT.length,
storage.getPathInfo(probe).getLength());
assertTrue(storage.deleteFile(probe));
- // The default-URI storage is bound to the local filesystem no matter what
the configuration
- // carries. This is the whole reason the path-aware overload has to be
used.
- assertEquals("file",
HoodieStorageUtils.getStorage(storage.getConf()).getScheme());
+ // Storage built from the default URI is bound to the local filesystem no
matter what the
+ // configuration carries, which is why the table's own path has to select
it.
+ assertEquals("file",
HoodieTestUtils.getLocalStorage(storage.getConf()).getScheme());
}
/**
@@ -308,7 +257,7 @@ class TestRollbackHelperV1 extends HoodieRollbackTestBase {
assertTrue(found.get(0).isPresent());
assertEquals(LOG_FILE_CONTENT.length, found.get(0).get().getLength());
- HoodieStorage defaultUriStorage =
HoodieStorageUtils.getStorage(storageConf);
+ HoodieStorage defaultUriStorage =
HoodieTestUtils.getLocalStorage(storageConf);
assertEquals("file", defaultUriStorage.getScheme());
assertThrows(
IllegalArgumentException.class,
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestEightToSevenDowngradeHandler.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestEightToSevenDowngradeHandler.java
index 49823f205c3d..675c044a917b 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestEightToSevenDowngradeHandler.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestEightToSevenDowngradeHandler.java
@@ -28,7 +28,6 @@ import org.apache.hudi.common.model.BootstrapIndexType;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
-import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieUpgradeDowngradeException;
@@ -66,7 +65,7 @@ import static
org.apache.hudi.common.table.HoodieTableConfig.PARTITION_FIELDS;
import static
org.apache.hudi.common.table.HoodieTableConfig.PAYLOAD_CLASS_NAME;
import static org.apache.hudi.common.table.HoodieTableConfig.RECORD_MERGE_MODE;
import static
org.apache.hudi.common.table.HoodieTableConfig.TABLE_METADATA_PARTITIONS;
-import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorage;
import static org.apache.hudi.metadata.MetadataPartitionType.COLUMN_STATS;
import static org.apache.hudi.metadata.MetadataPartitionType.FILES;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
@@ -132,7 +131,7 @@ class TestEightToSevenDowngradeHandler {
@Test
void testDowngradeMetadataPartitions() {
String baseTablePath = baseDir.toString();
- HoodieStorage hoodieStorage =
HoodieStorageUtils.getStorage(getDefaultStorageConf());
+ HoodieStorage hoodieStorage = getDefaultStorage();
StoragePath basePath = new StoragePath(baseTablePath);
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestCleanActionExecutor.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestCleanActionExecutor.java
index 752127448737..801e0cb31b32 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestCleanActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestCleanActionExecutor.java
@@ -62,6 +62,7 @@ import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_FILE_NAME
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.convertMetadataToByteArray;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
+import static org.apache.hudi.common.testutils.HoodieTestUtils.getLocalStorage;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
@@ -93,7 +94,7 @@ public class TestCleanActionExecutor {
when(mockHoodieTable.getMetaClient()).thenReturn(metaClient);
HoodieTableConfig tableConfig = new HoodieTableConfig();
when(metaClient.getTableConfig()).thenReturn(tableConfig);
- storage = spy(HoodieStorageUtils.getStorage(CONF));
+ storage = spy(getLocalStorage(CONF));
when(metaClient.getStorage()).thenReturn(storage);
when(mockHoodieTable.getStorage()).thenReturn(storage);
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/util/HoodieStorageUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/util/HoodieStorageUtils.java
index 4f6ed4a1dacb..c44ad8c5bc6c 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/util/HoodieStorageUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/util/HoodieStorageUtils.java
@@ -25,12 +25,14 @@ import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
+/**
+ * Factory for {@link HoodieStorage}.
+ *
+ * <p>Every overload takes the path the storage will be used for. {@link
HoodieStorage} binds one
+ * filesystem for its lifetime, and the filesystem is selected from the path's
scheme rather than
+ * from the configuration, so a handle built for one scheme cannot serve
another.
+ */
public class HoodieStorageUtils {
- public static final String DEFAULT_URI = "file:///";
-
- public static HoodieStorage getStorage(StorageConfiguration<?> conf) {
- return getStorage(DEFAULT_URI, conf);
- }
public static HoodieStorage getStorage(String basePath,
StorageConfiguration<?> conf) {
return getStorage(new StoragePath(basePath), conf);
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 bc7638d895b3..ae9deef4f7e5 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
@@ -92,7 +92,6 @@ import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
import static
org.apache.hudi.common.testutils.SchemaTestUtil.getSchemaFromResource;
-import static org.apache.hudi.common.util.HoodieStorageUtils.DEFAULT_URI;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -107,6 +106,11 @@ public class HoodieTestUtils {
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";
+ /**
+ * Base URI for storage that a test does not care about the location of.
Test-only: production
+ * code has no default filesystem, it resolves storage from the path it is
about to touch.
+ */
+ public static final String DEFAULT_URI = "file:///";
public static final InstantGenerator INSTANT_GENERATOR = new
DefaultInstantGenerator();
public static final TimelineFactory TIMELINE_FACTORY = new
DefaultTimelineFactory();
public static final InstantFileNameGenerator INSTANT_FILE_NAME_GENERATOR =
new DefaultInstantFileNameGenerator();
@@ -146,6 +150,16 @@ public class HoodieTestUtils {
return getStorage(DEFAULT_URI);
}
+ /**
+ * Storage on the local filesystem, carrying the given configuration.
+ *
+ * <p>Use when a test has a configuration to preserve but no path to resolve
from. Prefer
+ * {@link #getStorage(StoragePath)} whenever a path is available.
+ */
+ public static HoodieStorage getLocalStorage(StorageConfiguration<?> conf) {
+ return HoodieStorageUtils.getStorage(DEFAULT_URI, conf);
+ }
+
public static HoodieStorage getStorage(String path) {
return HoodieStorageUtils.getStorage(path, getDefaultStorageConf());
}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/NonLocalSchemeLocalFileSystem.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/NonLocalSchemeLocalFileSystem.java
new file mode 100644
index 000000000000..505d3c96800c
--- /dev/null
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/NonLocalSchemeLocalFileSystem.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.hadoop.fs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+
+import java.io.IOException;
+import java.net.URI;
+
+/**
+ * The local filesystem exposed under a scheme other than {@code file}, so a
test can tell apart
+ * storage resolved from a path (correct) and storage resolved from a default
{@code file:///} URI
+ * (wrong), the way s3a and gs do in production without needing a remote
object store.
+ *
+ * <p>{@link RawLocalFileSystem#pathToFile} keeps only the path component of a
URI, so a
+ * {@code <scheme>://<bucket>/tmp/x} path reads and writes the local file
{@code /tmp/x}. What this
+ * subclass changes is only the identity the filesystem reports, which is what
+ * {@link FileSystem#checkPath} validates every path against.
+ *
+ * <p>Register it per scheme on the test's configuration:
+ *
+ * <pre>
+ * conf.setClass("fs.s3a.impl", NonLocalSchemeLocalFileSystem.class,
FileSystem.class);
+ * </pre>
+ *
+ * A scheme is only safe to borrow when no real implementation claims it on
the module's test
+ * classpath.
+ */
+public class NonLocalSchemeLocalFileSystem extends RawLocalFileSystem {
+
+ /**
+ * Answer for {@link #getUri()} until {@link #initialize} supplies the real
one. The superclass
+ * constructor calls {@code getUri()}, which runs before any instance field
of this subclass is
+ * assigned, so the fallback has to be a static. The scheme is a
placeholder: it is replaced on
+ * the first {@code initialize} call, and no path is validated against it
before then.
+ */
+ private static final URI UNINITIALIZED_URI = URI.create("uninitialized:///");
+
+ private URI uri;
+
+ @Override
+ public void initialize(URI name, Configuration conf) throws IOException {
+ super.initialize(name, conf);
+ this.uri = URI.create(name.getScheme() + "://"
+ + (name.getAuthority() == null ? "" : name.getAuthority()));
+ setWorkingDirectory(new Path(this.uri.toString() + Path.SEPARATOR));
+ }
+
+ @Override
+ public URI getUri() {
+ return uri == null ? UNINITIALIZED_URI : uri;
+ }
+
+ @Override
+ public String getScheme() {
+ return getUri().getScheme();
+ }
+
+ /**
+ * The superclass qualifies the process working directory against {@link
#getUri()} from its own
+ * constructor. Skip that: this filesystem does not know its real URI yet,
and every path a test
+ * gives it is absolute, so the working directory is never consulted.
+ */
+ @Override
+ protected Path getInitialWorkingDirectory() {
+ return new Path(System.getProperty("user.dir"));
+ }
+}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieHadoopIOFactory.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieHadoopIOFactory.java
index ae3c5f6d7657..053d9c41e66a 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieHadoopIOFactory.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/hadoop/TestHoodieHadoopIOFactory.java
@@ -35,8 +35,8 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
+import static org.apache.hudi.common.testutils.HoodieTestUtils.DEFAULT_URI;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
-import static org.apache.hudi.common.util.HoodieStorageUtils.DEFAULT_URI;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReader.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReader.scala
index f77fb61a5c8c..2074dee48d01 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReader.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReader.scala
@@ -25,7 +25,6 @@ import org.apache.hudi.common.util.HoodieStorageUtils
import org.apache.hudi.io.SeekableDataInputStream
import org.apache.hudi.storage.{HoodieStorage, StorageConfiguration,
StoragePath}
-import org.apache.spark.TaskContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, Row}
@@ -89,12 +88,14 @@ import scala.collection.mutable.ArrayBuffer
* <li>Tune maxGapBytes based on your data access patterns</li>
* </ul>
*
- * @param storage HoodieStorage instance for file I/O
+ * @param resolveStorage Supplies the HoodieStorage for a blob reference. A
HoodieStorage is bound
+ * to the filesystem of the path it is built with, and a
reference is only
+ * known once its row arrives, so this is called per
read rather than once.
* @param maxGapBytes Maximum gap between ranges to consider for batching
(default: 4KB)
* @param lookaheadRows Number of rows to buffer for batch detection
(default: 50)
*/
class BatchedBlobReader(
- storage: HoodieStorage,
+ resolveStorage: StoragePath => HoodieStorage,
maxGapBytes: Int = 4096,
lookaheadRows: Int = 50) {
@@ -387,9 +388,11 @@ class BatchedBlobReader(
outputSchema: StructType)
(implicit builder: RowBuilder[R]): RowResult[R] = {
+ var storage: HoodieStorage = null
var inputStream: InputStream = null
try {
val path = new StoragePath(rowInfo.filePath)
+ storage = resolveStorage(path)
inputStream = storage.open(path)
val buffer = inputStream.readAllBytes()
@@ -405,6 +408,7 @@ class BatchedBlobReader(
logger.warn(s"Error closing stream for ${rowInfo.filePath}", e)
}
}
+ closeStorage(storage, rowInfo.filePath)
}
}
@@ -426,10 +430,13 @@ class BatchedBlobReader(
outputSchema: StructType)
(implicit builder: RowBuilder[R]): Seq[RowResult[R]] = {
+ var storage: HoodieStorage = null
var inputStream: SeekableDataInputStream = null
try {
// Get or open file handle
- inputStream = storage.openSeekable(new StoragePath(range.filePath),
false)
+ val path = new StoragePath(range.filePath)
+ storage = resolveStorage(path)
+ inputStream = storage.openSeekable(path, false)
// Seek to start offset
inputStream.seek(range.startOffset)
@@ -477,6 +484,23 @@ class BatchedBlobReader(
logger.warn(s"Error closing input stream for ${range.filePath}", e)
}
}
+ closeStorage(storage, range.filePath)
+ }
+ }
+
+ /**
+ * Storage is resolved per read, so the read that resolved it closes it.
HoodieHadoopStorage
+ * treats this as a no-op because it does not own the cached Hadoop
filesystem, but another
+ * hoodie.storage.class implementation may hold resources of its own.
+ */
+ private def closeStorage(storage: HoodieStorage, filePath: String): Unit = {
+ if (storage != null) {
+ try {
+ storage.close()
+ } catch {
+ case e: Exception =>
+ logger.warn(s"Error closing storage for $filePath", e)
+ }
}
}
}
@@ -691,9 +715,9 @@ object BatchedBlobReader {
// Apply mapPartitions
val result = df.mapPartitions { partition =>
- // Create storage and reader for this partition
- val storage = HoodieStorageUtils.getStorage(broadcastConf.value)
- val reader = new BatchedBlobReader(storage, maxGapBytes, lookaheadSize)
+ // Create reader for this partition
+ val reader = new BatchedBlobReader(
+ HoodieStorageUtils.getStorage(_, broadcastConf.value), maxGapBytes,
lookaheadSize)
// Import implicit instances for Row
import RowAccessor.rowAccessor
@@ -701,7 +725,6 @@ object BatchedBlobReader {
// Process partition
val iter = reader.processPartition[Row](partition, structColIdx,
outputSchema)
- TaskContext.get().addTaskCompletionListener[Unit](_ => storage.close())
iter
} (sparkAdapter.getCatalystExpressionUtils.getEncoder(outputSchema))
@@ -751,16 +774,14 @@ object BatchedBlobReader {
// Process partitions using InternalRow type classes
rdd.mapPartitions { partition =>
- val storage = HoodieStorageUtils.getStorage(broadcastConf.value)
-
- val reader = new BatchedBlobReader(storage, maxGapBytes, lookaheadSize)
+ val reader = new BatchedBlobReader(
+ HoodieStorageUtils.getStorage(_, broadcastConf.value), maxGapBytes,
lookaheadSize)
// Import implicit instances for InternalRow
import RowAccessor.internalRowAccessor
import RowBuilder.internalRowBuilder
val iter = reader.processPartition[InternalRow](partition, structColIdx,
outputSchema)
- TaskContext.get().addTaskCompletionListener[Unit](_ => storage.close())
iter
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/TestIncrementalQueryWithArchivedInstants.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/TestIncrementalQueryWithArchivedInstants.scala
index 64cf0371dd36..5036b207a438 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/TestIncrementalQueryWithArchivedInstants.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/TestIncrementalQueryWithArchivedInstants.scala
@@ -22,7 +22,7 @@ package org.apache
import org.apache.hudi.DataSourceWriteOptions.RECORD_MERGE_MODE
import org.apache.hudi.common.config.RecordMergeMode
import org.apache.hudi.common.table.HoodieTableMetaClient
-import org.apache.hudi.common.util.HoodieStorageUtils
+import org.apache.hudi.common.testutils.HoodieTestUtils
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness
@@ -86,7 +86,7 @@ class TestIncrementalQueryWithArchivedInstants extends
SparkClientFunctionalTest
val storageConfiguration = new HadoopStorageConfiguration(false)
val metaClient = HoodieTableMetaClient.builder().setBasePath(tblPath)
- .setStorage(HoodieStorageUtils.getStorage(storageConfiguration)).build()
+
.setStorage(HoodieTestUtils.getLocalStorage(storageConfiguration)).build()
val instants = metaClient.getArchivedTimeline().getInstants
// There are at least one archived instants.
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/BlobTestHelpers.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/BlobTestHelpers.scala
index b94229aa478e..d88c2498389f 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/BlobTestHelpers.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/BlobTestHelpers.scala
@@ -20,13 +20,17 @@
package org.apache.hudi.blob
import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaType}
+import org.apache.hudi.hadoop.fs.NonLocalSchemeLocalFileSystem
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.FileSystem
import org.apache.spark.sql.{Column, DataFrame, SparkSession}
import org.apache.spark.sql.functions.{lit, struct}
import org.apache.spark.sql.types._
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
import java.io.File
+import java.net.URI
import java.nio.file.{Files, Path}
object BlobTestHelpers {
@@ -71,6 +75,37 @@ object BlobTestHelpers {
).as(name, blobMetadata)
}
+ /**
+ * Exposes the local filesystem under the given non-local schemes, so a test
can tell apart storage
+ * resolved from a path and storage resolved from a default file:/// URI
without a remote object
+ * store.
+ *
+ * Hadoop caches filesystems statically, keyed by scheme and authority
rather than by
+ * configuration, so a borrowed impl would outlive the test for the whole
fork and the next test to
+ * use that scheme would silently read local files. Both the configuration
and the cache are put
+ * back afterwards.
+ */
+ def withBorrowedSchemes[T](hadoopConf: Configuration, authorities:
Seq[String], schemes: String*)(body: => T): T = {
+ val previous = schemes.map(scheme => scheme ->
Option(hadoopConf.get(s"fs.$scheme.impl"))).toMap
+ schemes.foreach(scheme =>
+ hadoopConf.setClass(s"fs.$scheme.impl",
classOf[NonLocalSchemeLocalFileSystem], classOf[FileSystem]))
+ try {
+ body
+ } finally {
+ for (scheme <- schemes; authority <- authorities) {
+ // FileSystem.close removes the instance from the static cache.
+ try FileSystem.get(new URI(s"$scheme://$authority"),
hadoopConf).close()
+ catch { case _: Exception => }
+ }
+ schemes.foreach { scheme =>
+ previous(scheme) match {
+ case Some(impl) => hadoopConf.set(s"fs.$scheme.impl", impl)
+ case None => hadoopConf.unset(s"fs.$scheme.impl")
+ }
+ }
+ }
+ }
+
def createTestFile(tempDir: Path, name: String, size: Int): String = {
val file = new File(tempDir.toString, name)
val bytes = (0 until size).map(i => (i % 256).toByte).toArray
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/CountingHoodieStorage.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/CountingHoodieStorage.scala
new file mode 100644
index 000000000000..286f02db9e5e
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/CountingHoodieStorage.scala
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.blob
+
+import org.apache.hudi.storage.{StorageConfiguration, StoragePath}
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage
+
+import java.util.concurrent.atomic.AtomicInteger
+
+/**
+ * A hoodie.storage.class that counts construction and close, so a test can
assert every storage a
+ * reader resolves is also closed. HoodieHadoopStorage.close is deliberately a
no-op because it does
+ * not own the cached Hadoop filesystem, which is exactly why a no-op cannot
stand in for proof that
+ * close is called.
+ *
+ * The counters are static and the tests using them run Spark in-process, so
executor-side
+ * construction is visible to the assertions.
+ */
+class CountingHoodieStorage(path: StoragePath, conf: StorageConfiguration[_])
+ extends HoodieHadoopStorage(path, conf) {
+
+ CountingHoodieStorage.constructed.incrementAndGet()
+
+ override def close(): Unit = {
+ CountingHoodieStorage.closed.incrementAndGet()
+ super.close()
+ }
+}
+
+object CountingHoodieStorage {
+ val constructed = new AtomicInteger(0)
+ val closed = new AtomicInteger(0)
+
+ def reset(): Unit = {
+ constructed.set(0)
+ closed.set(0)
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBatchedBlobReader.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBatchedBlobReader.scala
index 12f4496cc377..e9da416f5b76 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBatchedBlobReader.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBatchedBlobReader.scala
@@ -20,9 +20,16 @@
package org.apache.hudi.blob
import org.apache.hudi.blob.BlobTestHelpers._
+import org.apache.hudi.common.config.HoodieStorageConfig
import org.apache.hudi.common.schema.HoodieSchema
+import org.apache.hudi.common.testutils.HoodieTestUtils
+import org.apache.hudi.common.util.HoodieStorageUtils
+import org.apache.hudi.hadoop.fs.{HadoopFSUtils, NonLocalSchemeLocalFileSystem}
+import org.apache.hudi.storage.{StorageConfiguration, StoragePath}
import org.apache.hudi.testutils.HoodieClientTestBase
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.FileSystem
import org.apache.spark.SparkException
import org.apache.spark.sql.Row
import org.apache.spark.sql.functions._
@@ -454,4 +461,147 @@ class TestBatchedBlobReader extends HoodieClientTestBase {
assertTrue(thrown.getCause.isInstanceOf[IllegalArgumentException])
assertTrue(thrown.getCause.getMessage.contains("Overlapping blob ranges
detected"))
}
+
+ /**
+ * Blob references are absolute paths carried in row data, so the filesystem
a partition must read
+ * is not known until the rows arrive. These tests put the referenced files
behind an object-store
+ * scheme, which is where storage resolved from a default {@code file:///}
URI fails with
+ * {@code IllegalArgumentException: Wrong FS: s3a://..., expected: file:///}.
+ */
+ @Test
+ def testRangedReadOfAReferenceOnANonLocalScheme(): Unit = {
+ val localPath = createTestFile(tempDir, "ranged-non-local.bin", 10000)
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (onScheme("s3a", localPath), 0L, 100L),
+ (onScheme("s3a", localPath), 100L, 100L)
+ )).toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"),
col("offset"), col("length")))
+ .select("offset", "data")
+ .coalesce(1)
+
+ val results = BatchedBlobReader.readBatched(inputDF,
nonLocalSchemeStorageConf).orderBy("offset").collect()
+
+ assertEquals(2, results.length)
+ results.zipWithIndex.foreach { case (row, i) =>
+ val data = row.getAs[Array[Byte]]("data")
+ assertEquals(100, data.length)
+ assertBytesContent(data, expectedOffset = i * 100)
+ }
+ }
+
+ @Test
+ def testWholeFileReadOfAReferenceOnANonLocalScheme(): Unit = {
+ val localPath = createTestFile(tempDir, "whole-non-local.bin", 512)
+ val inputDF = sparkSession.createDataFrame(Seq(Tuple1(onScheme("gs",
localPath))))
+ .toDF("external_path")
+ .withColumn("data", wholeFileBlobStructCol("data", col("external_path")))
+ .select("data")
+ .coalesce(1)
+
+ val results = BatchedBlobReader.readBatched(inputDF,
nonLocalSchemeStorageConf).collect()
+
+ assertEquals(1, results.length)
+ val data = results(0).getAs[Array[Byte]]("data")
+ assertEquals(512, data.length)
+ assertBytesContent(data)
+ }
+
+ /**
+ * One partition referencing two buckets on the same scheme. Hadoop keys its
filesystem cache by
+ * scheme and authority and FileSystem.checkPath validates both, so a reader
that resolved storage
+ * once would serve the second bucket through the first bucket's handle and
fail with
+ * Wrong FS: s3a://bucket-b/..., expected: s3a://bucket-a.
+ */
+ @Test
+ def testReferencesInTwoBucketsInOnePartition(): Unit = {
+ val firstPath = createTestFile(tempDir, "two-bucket-first.bin", 1000)
+ val secondPath = createTestFile(tempDir, "two-bucket-second.bin", 1000)
+ val inputDF = sparkSession.createDataFrame(Seq(
+ ("a", s"s3a://bucket-a$firstPath", 0L, 100L),
+ ("b", s"s3a://bucket-b$secondPath", 0L, 100L)
+ )).toDF("bucket", "external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"),
col("offset"), col("length")))
+ .select("bucket", "data")
+ .coalesce(1)
+
+ val results = BatchedBlobReader.readBatched(inputDF,
nonLocalSchemeStorageConf).orderBy("bucket").collect()
+
+ assertEquals(2, results.length)
+ assertEquals(Seq("a", "b"), results.map(_.getAs[String]("bucket")).toSeq)
+ results.foreach { row =>
+ val data = row.getAs[Array[Byte]]("data")
+ assertEquals(100, data.length)
+ assertBytesContent(data)
+ }
+ }
+
+ /**
+ * Every storage the reader resolves is closed. HoodieHadoopStorage.close is
a no-op because it
+ * does not own the cached Hadoop filesystem, but hoodie.storage.class is
pluggable and another
+ * implementation may hold resources, so the reader must not leak the
handles it creates.
+ */
+ @Test
+ def testResolvedStorageIsClosed(): Unit = {
+ val localPath = createTestFile(tempDir, "counted-close.bin", 1000)
+ val conf = nonLocalSchemeStorageConf
+ conf.set(HoodieStorageConfig.HOODIE_STORAGE_CLASS.key,
classOf[CountingHoodieStorage].getName)
+
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (onScheme("s3a", localPath), 0L, 100L),
+ (onScheme("s3a", localPath), 500L, 100L)
+ )).toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"),
col("offset"), col("length")))
+ .select("offset", "data")
+ .coalesce(1)
+
+ CountingHoodieStorage.reset()
+ val results = BatchedBlobReader.readBatched(inputDF, conf).collect()
+ assertEquals(2, results.length)
+
+ val constructed = CountingHoodieStorage.constructed.get()
+ assertTrue(constructed > 0, "the reader should have resolved storage at
least once")
+ assertEquals(constructed, CountingHoodieStorage.closed.get(),
+ s"every resolved storage must be closed, constructed $constructed")
+ }
+
+ /**
+ * Harness sanity, so none of the tests above can pass for the wrong reason:
the borrowed schemes
+ * must reach the local file the test wrote, and must not be silently
rewritten to {@code file}.
+ */
+ @Test
+ def testBorrowedSchemesReachLocalFiles(): Unit = {
+ val localPath = createTestFile(tempDir, "harness-sanity.bin", 256)
+ val conf = nonLocalSchemeStorageConf
+ Seq("s3a", "gs").foreach { scheme =>
+ val path = new StoragePath(onScheme(scheme, localPath))
+ val storage = HoodieStorageUtils.getStorage(path, conf)
+ try {
+ assertEquals(scheme, storage.getScheme)
+ assertEquals(256, storage.getPathInfo(path).getLength)
+ } finally {
+ storage.close()
+ }
+ }
+ val localStorage = HoodieTestUtils.getLocalStorage(conf)
+ try {
+ assertEquals("file", localStorage.getScheme)
+ } finally {
+ localStorage.close()
+ }
+ }
+
+ private def onScheme(scheme: String, localPath: String): String =
s"$scheme://test-bucket$localPath"
+
+ /**
+ * The table's storage configuration with the local filesystem also exposed
under two object-store
+ * schemes. Neither scheme has a real implementation on this module's test
classpath, so nothing
+ * else claims them.
+ */
+ private def nonLocalSchemeStorageConf: StorageConfiguration[_] = {
+ val hadoopConf = new
Configuration(storageConf.unwrapAs(classOf[Configuration]))
+ Seq("s3a", "gs").foreach { scheme =>
+ hadoopConf.setClass(s"fs.$scheme.impl",
classOf[NonLocalSchemeLocalFileSystem], classOf[FileSystem])
+ }
+ HadoopFSUtils.getStorageConf(hadoopConf)
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestReadBlobSQL.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestReadBlobSQL.scala
index 533c9589e338..e247e9758f2f 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestReadBlobSQL.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestReadBlobSQL.scala
@@ -103,6 +103,61 @@ class TestReadBlobSQL extends HoodieClientTestBase {
}
}
+ /**
+ * The same out-of-line read, but with the blob reference on an object-store
scheme while the table
+ * stays local. read_blob() reaches BatchedBlobReader through
BatchedBlobReadExec and processRDD,
+ * which is a separate call site from the readBatched path covered in
TestBatchedBlobReader, so
+ * storage resolved from a default file:/// URI fails here with
+ * IllegalArgumentException: Wrong FS: s3a://..., expected: file:///.
+ */
+ @Test
+ def testReadOutOfLineBlobReferenceOnANonLocalScheme(): Unit = {
+ // BatchedBlobReaderStrategy builds the storage configuration from the
session's Hadoop conf,
+ // so the borrowed scheme has to be registered there rather than on a conf
the test holds.
+ withBorrowedSchemes(sparkSession.sparkContext.hadoopConfiguration,
Seq("test-bucket"), "s3a") {
+ val extFile = createTestFile(tempDir, "non-local-sql.bin", 10000)
+ val reference = s"s3a://test-bucket$extFile"
+ val tablePath = s"$tempDir/hudi_blob_table_non_local"
+
+ val rawDf = sparkSession.createDataFrame(Seq(
+ (1, "rec1", reference, 0L, 100L),
+ (2, "rec2", reference, 100L, 100L)
+ )).toDF("id", "name", "external_path", "offset", "length")
+ .withColumn("file_info",
+ blobStructCol("file_info", col("external_path"), col("offset"),
col("length")))
+ .select("id", "name", "file_info")
+
+ val canonicalSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = false),
+ StructField("name", StringType, nullable = true),
+ StructField("file_info", BlobType().asInstanceOf[StructType], nullable
= true, blobMetadata)
+ ))
+ sparkSession.createDataFrame(rawDf.rdd,
canonicalSchema).write.format("hudi")
+ .option("hoodie.table.name", "blob_test_non_local")
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.operation", "bulk_insert")
+ .mode("overwrite")
+ .save(tablePath)
+
+ sparkSession.read.format("hudi").load(tablePath)
+ .createOrReplaceTempView("hudi_blob_non_local_view")
+
+ val result = sparkSession.sql("""
+ SELECT id, read_blob(file_info) AS data
+ FROM hudi_blob_non_local_view
+ ORDER BY id
+ """).collect()
+
+ assertEquals(2, result.length)
+ result.zipWithIndex.foreach { case (row, idx) =>
+ assertEquals(idx + 1, row.getInt(0))
+ val bytes = row.getAs[Array[Byte]]("data")
+ assertEquals(100, bytes.length)
+ assertBytesContent(bytes, expectedOffset = idx * 100)
+ }
+ }
+ }
+
@Test
def testBasicReadBlobSQL(): Unit = {
val filePath = createTestFile(tempDir, "basic.bin", 10000)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/blob/TestBatchedBlobReaderMerge.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/blob/TestBatchedBlobReaderMerge.scala
index 219f487a3c66..29aa75c89dbe 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/blob/TestBatchedBlobReaderMerge.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/blob/TestBatchedBlobReaderMerge.scala
@@ -27,7 +27,7 @@ import org.junit.jupiter.api.Test
* Direct unit tests for the merge algorithm in {@link BatchedBlobReader}.
*
* These tests bypass Spark and storage entirely: they construct a reader with
- * a null storage (the merge methods do not touch I/O) and call mergeRanges /
+ * a null storage resolver (the merge methods do not touch I/O) and call
mergeRanges /
* identifyConsecutiveRanges with crafted RowInfo inputs to assert the
* structure of the merged output (counts, boundaries, ordering, grouping).
*
@@ -38,7 +38,7 @@ import org.junit.jupiter.api.Test
class TestBatchedBlobReaderMerge {
private def reader(maxGapBytes: Int = 4096) =
- new BatchedBlobReader(storage = null, maxGapBytes = maxGapBytes,
lookaheadRows = 50)
+ new BatchedBlobReader(resolveStorage = null, maxGapBytes = maxGapBytes,
lookaheadRows = 50)
private def row(filePath: String, offset: Long, length: Long, index: Long =
0L): RowInfo[Row] =
RowInfo[Row](
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestUtilHelpers.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestUtilHelpers.java
index 1b6e95bafa50..84c40aade4aa 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestUtilHelpers.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestUtilHelpers.java
@@ -20,7 +20,6 @@ package org.apache.hudi.utilities;
import org.apache.hudi.client.transaction.lock.FileSystemBasedLockProvider;
import org.apache.hudi.common.config.TypedProperties;
-import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieLockConfig;
import org.apache.hudi.config.HoodieWriteConfig;
@@ -46,7 +45,7 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
-import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorage;
import static
org.apache.hudi.testutils.HoodieClientTestUtils.getSparkConfForTest;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -83,7 +82,7 @@ public class TestUtilHelpers {
sparkSession,
new HoodieStreamerMetrics(
HoodieWriteConfig.newBuilder().withPath("mypath").build(),
- HoodieStorageUtils.getStorage(getDefaultStorageConf())),
+ getDefaultStorage()),
new DefaultStreamContext(new SchemaTestProvider(typedProperties),
Option.empty()));
assertTrue(source instanceof AvroKafkaSource);
}
@@ -100,7 +99,7 @@ public class TestUtilHelpers {
sparkSession,
new HoodieStreamerMetrics(
HoodieWriteConfig.newBuilder().withPath("mypath").build(),
- HoodieStorageUtils.getStorage(getDefaultStorageConf())),
+ getDefaultStorage()),
new DefaultStreamContext(new SchemaTestProvider(typedProperties),
Option.empty())));
// We expect two constructors to complain about this error.
assertEquals(2, e.getSuppressed().length);
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java
index 4b8e975a5fb3..7cc3d3a8c19d 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java
@@ -42,7 +42,6 @@ import
org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.ConfigUtils;
-import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.ObjectSizeCalculator;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.ExternalSpillableMap;
@@ -105,7 +104,7 @@ import static
org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION
import static
org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.DEFAULT_PARTITION_PATHS;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
-import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorage;
import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
import static
org.apache.hudi.utilities.sources.helpers.IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -175,7 +174,7 @@ public class TestHoodieIncrSource extends
SparkClientFunctionalTestHarness {
HoodieStreamerMetrics metrics = new
HoodieStreamerMetrics(HoodieWriteConfig.newBuilder()
.withProperties(writeConfig.getProps()).withMetricsConfig(HoodieMetricsConfig.newBuilder()
.on(true).build()).build(),
- HoodieStorageUtils.getStorage(getDefaultStorageConf()));
+ getDefaultStorage());
// Reset mock to clear any previous invocations
// reset(metrics);
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerMetrics.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerMetrics.java
index 1fa85ad8cfc2..0fd69b47b549 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerMetrics.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerMetrics.java
@@ -20,7 +20,6 @@
package org.apache.hudi.utilities.streamer;
import org.apache.hudi.common.config.metrics.HoodieMetricsConfig;
-import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerMetrics;
@@ -30,7 +29,7 @@ import org.junit.jupiter.api.Test;
import java.util.concurrent.TimeUnit;
-import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorage;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -46,7 +45,7 @@ public class TestHoodieStreamerMetrics {
.withReporterType("INMEMORY")
.build();
HoodieStreamerMetrics metrics = new HoodieStreamerMetrics(
- metricsConfig, HoodieStorageUtils.getStorage(getDefaultStorageConf()));
+ metricsConfig, getDefaultStorage());
Timer.Context timerContext = metrics.getErrorTableWriteTimerContext();
Thread.sleep(TimeUnit.SECONDS.toMillis(1));
long duration = timerContext.stop();
@@ -64,7 +63,7 @@ public class TestHoodieStreamerMetrics {
.withReporterType("INMEMORY")
.build();
HoodieStreamerMetrics metrics = new HoodieStreamerMetrics(
- metricsConfig, HoodieStorageUtils.getStorage(getDefaultStorageConf()));
+ metricsConfig, getDefaultStorage());
Timer.Context timerContext = metrics.getErrorTableWriteTimerContext();
assertNull(timerContext);
metrics.updateErrorTableCommitDuration(0L);
@@ -79,7 +78,7 @@ public class TestHoodieStreamerMetrics {
.withReporterType("INMEMORY")
.build();
HoodieStreamerMetrics metrics = new HoodieStreamerMetrics(
- metricsConfig, HoodieStorageUtils.getStorage(getDefaultStorageConf()));
+ metricsConfig, getDefaultStorage());
metrics.emitStreamerJobSuccessMetrics();
MetricRegistry registry = metrics.getMetrics().getRegistry();
assertEquals(1, registry.getGauges().size());
@@ -95,7 +94,7 @@ public class TestHoodieStreamerMetrics {
.withReporterType("INMEMORY")
.build();
HoodieStreamerMetrics metrics = new HoodieStreamerMetrics(
- metricsConfig, HoodieStorageUtils.getStorage(getDefaultStorageConf()));
+ metricsConfig, getDefaultStorage());
metrics.emitStreamerJobFailedMetrics();
MetricRegistry registry = metrics.getMetrics().getRegistry();
assertEquals(1, registry.getGauges().size());
@@ -111,7 +110,7 @@ public class TestHoodieStreamerMetrics {
.withReporterType("INMEMORY")
.build();
HoodieStreamerMetrics metrics = new HoodieStreamerMetrics(
- metricsConfig, HoodieStorageUtils.getStorage(getDefaultStorageConf()));
+ metricsConfig, getDefaultStorage());
// Should not throw when metrics are disabled
metrics.emitStreamerJobSuccessMetrics();
metrics.emitStreamerJobFailedMetrics();
@@ -126,7 +125,7 @@ public class TestHoodieStreamerMetrics {
.withReporterType("INMEMORY")
.build();
HoodieDeltaStreamerMetrics metrics = new HoodieDeltaStreamerMetrics(
- metricsConfig, HoodieStorageUtils.getStorage(getDefaultStorageConf()));
+ metricsConfig, getDefaultStorage());
metrics.emitStreamerJobSuccessMetrics();
assertEquals(".deltastreamer.success",
metrics.getMetrics().getRegistry().getGauges().firstKey());
@@ -136,7 +135,7 @@ public class TestHoodieStreamerMetrics {
.withMetricsConfig(HoodieMetricsConfig.newBuilder().on(true).withReporterType("INMEMORY").build())
.build();
HoodieDeltaStreamerMetrics metricsFromWriteConfig = new
HoodieDeltaStreamerMetrics(
- writeConfig, HoodieStorageUtils.getStorage(getDefaultStorageConf()));
+ writeConfig, getDefaultStorage());
metricsFromWriteConfig.emitStreamerJobFailedMetrics();
assertEquals(".deltastreamer.failure",
metricsFromWriteConfig.getMetrics().getRegistry().getGauges().firstKey());
