This is an automated email from the ASF dual-hosted git repository.
stevel pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hadoop-api-shim.git
commit fc3b60ee4479e812ead37a47f8f4837307db802e
Author: Steve Loughran <ste...@cloudera.com>
AuthorDate: Fri Jul 28 17:29:50 2023 +0100
PARQUET-2171. Vector IO through self-contained shim classes.
Takes the vector IO shim code from the hadoop-shim library and
reimplements in through the DynMethods class.
Needs
* lower level tests of behaviour
* review of exception handling/unwrapping.
---
.../fs/shim/impl/FSDataInputStreamShimImpl.java | 10 ++-
.../hadoop/fs/shim/impl/FileRangeBridge.java | 81 ++++++++++++++--------
.../hadoop/fs/shim/impl/FileSystemShimImpl.java | 5 +-
.../hadoop/fs/shim/impl/ShimReflectionSupport.java | 10 +--
4 files changed, 64 insertions(+), 42 deletions(-)
diff --git
a/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FSDataInputStreamShimImpl.java
b/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FSDataInputStreamShimImpl.java
index f1b535e..45df066 100644
---
a/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FSDataInputStreamShimImpl.java
+++
b/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FSDataInputStreamShimImpl.java
@@ -25,7 +25,6 @@ import java.nio.ByteBuffer;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Function;
import java.util.function.IntFunction;
import java.util.stream.Collectors;
@@ -103,15 +102,14 @@ public class FSDataInputStreamShimImpl
public FSDataInputStreamShimImpl(
final FSDataInputStream instance) {
super(FSDataInputStream.class, instance);
- byteBufferPositionedRead = loadInvocation(getClazz(), READ,
- Integer.class,
+ byteBufferPositionedRead = loadInvocation(getClazz(), Integer.class, READ,
Long.class, ByteBuffer.class);
boolean bbrb = instance.hasCapability(PREADBYTEBUFFER)
&& byteBufferPositionedRead.available();
if (bbrb) {
byteBufferPositionedReadFully = loadInvocation(getClazz(),
- READ_FULLY, Void.class, Long.class, ByteBuffer.class);
+ Void.class, READ_FULLY, Long.class, ByteBuffer.class);
isByteBufferPositionedReadAvailable = new AtomicBoolean(true);
} else {
byteBufferPositionedReadFully = unavailable(READ_FULLY);
@@ -122,8 +120,8 @@ public class FSDataInputStreamShimImpl
isByteBufferReadableAvailable = new AtomicBoolean(
instance.getWrappedStream() instanceof ByteBufferReadable);
if (FILE_RANGE_BRIDGE.bridgeAvailable()) {
- readVectored = loadInvocation(getClazz(), READ_VECTORED,
- Void.class, List.class, Function.class);
+ readVectored = loadInvocation(getClazz(), Void.class, READ_VECTORED,
+ List.class, IntFunction.class);
} else {
readVectored = unavailable(READ_VECTORED);
}
diff --git
a/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FileRangeBridge.java
b/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FileRangeBridge.java
index 1cc7afd..04c118b 100644
---
a/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FileRangeBridge.java
+++
b/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FileRangeBridge.java
@@ -18,8 +18,6 @@
package org.apache.hadoop.fs.shim.impl;
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
@@ -29,7 +27,6 @@ import org.slf4j.LoggerFactory;
import org.apache.hadoop.fs.shim.api.VectorFileRange;
import static java.util.Objects.requireNonNull;
-import static org.apache.hadoop.fs.shim.impl.ShimReflectionSupport.ctor;
import static
org.apache.hadoop.fs.shim.impl.ShimReflectionSupport.loadInvocation;
/**
@@ -38,15 +35,27 @@ import static
org.apache.hadoop.fs.shim.impl.ShimReflectionSupport.loadInvocatio
public final class FileRangeBridge {
private static final Logger LOG =
LoggerFactory.getLogger(FileRangeBridge.class);
- public static final String CLASSNAME =
"org.apache.hadoop.fs.impl.FileRangeImpl";
+ /**
+ * Name of the {@code FileRange} interface.
+ */
+ public static final String CLASSNAME = "org.apache.hadoop.fs.FileRange";
- private final Class<?> fileRangeClass;
+ /**
+ * Class of the interface {@link #CLASSNAME}, if loaded.
+ * This can resolve all methods in this interface and super-interfaces,
+ * including static ones.
+ */
+ private final Class<?> fileRangeInterface;
private final Invocation<Long> _getOffset;
private final Invocation<Integer> _getLength;
private final Invocation<CompletableFuture<ByteBuffer>> _getData;
private final Invocation<Void> _setData;
private final Invocation<Object> _getReference;
- private final Constructor<?> newFileRange;
+
+ /**
+ * new FileRange(long, long, Object)
+ */
+ private final Invocation<Object> createFileRange;
/**
* Constructor.
@@ -61,59 +70,67 @@ public final class FileRangeBridge {
LOG.debug("No {}", CLASSNAME);
cl = null;
}
- fileRangeClass = cl;
+ fileRangeInterface = cl;
// class found, so load the methods
- _getOffset = loadInvocation(fileRangeClass, "getOffset", Long.class);
- _getLength= loadInvocation(fileRangeClass, "getLength", Integer.class);
- _getData = loadInvocation(fileRangeClass, "getData", null);
- _setData = loadInvocation(fileRangeClass, "setData", Void.class,
CompletableFuture.class);
- _getReference = loadInvocation(fileRangeClass, "getReference",
Object.class);
+ _getOffset = loadInvocation(fileRangeInterface, Long.class, "getOffset");
+ _getLength = loadInvocation(fileRangeInterface, Integer.class,
"getLength");
+ _getData = loadInvocation(fileRangeInterface, null, "getData");
+ _setData = loadInvocation(fileRangeInterface, Void.class, "setData",
CompletableFuture.class);
+ _getReference = loadInvocation(fileRangeInterface, Object.class,
"getReference");
+
+ // static interface method to create an instance.
+ createFileRange = loadInvocation(fileRangeInterface, Object.class,
"createFileRange", Long.class,
+ Integer.class, Object.class);
- newFileRange = ctor(fileRangeClass, Long.class, Integer.class,
Object.class);
}
/**
* Is the bridge available.
+ *
* @return true iff the bridge is present.
*/
public boolean bridgeAvailable() {
- return fileRangeClass != null;
+ return fileRangeInterface != null;
}
/**
* Get the file range class.
+ *
* @return the file range implementation class, if present.
*/
- public Class<?> getFileRangeClass() {
- return fileRangeClass;
+ public Class<?> getFileRangeInterface() {
+ return fileRangeInterface;
}
/**
* Instantiate.
+ *
+ * @param offset offset in file
+ * @param length length of data to read.
+ * @param reference nullable reference to store in the range.
* @return a VectorFileRange wrapping a FileRange
+ *
* @throws RuntimeException if the range cannot be instantiated
* @throws IllegalStateException if the API is not available.
*/
- public WrappedFileRange newWrappedFileRange(long offset, int length) {
+ public VectorFileRange createFileRange(long offset, int length, final Object
reference) {
Preconditions.checkState(bridgeAvailable(), "FileRange not available");
- try {
- return new WrappedFileRange(newFileRange.newInstance(offset, length));
- } catch (InstantiationException | IllegalAccessException |
InvocationTargetException e) {
- throw new RuntimeException("failed to instantiate a FileRange object: "
+ e,
- e);
- }
+ return new WrappedFileRange(createFileRange.invokeUnchecked(null, offset,
length, reference));
}
/**
- * Convert a range to an instance of FileRange
- * @param in input
+ * Convert a range to an instance of FileRange.
+ * The offset, length and reference of the input range
+ * all passed into the createFileRange() method.
+ * @param range input range
+ *
* @return a converted instance
*/
- public Object toFileRange(VectorFileRange in) {
+ public Object toFileRange(VectorFileRange range) {
// create a new wrapped file range, fill in and then
// get the instance
- final WrappedFileRange wfr = newWrappedFileRange(in.getOffset(),
in.getLength());
- wfr.setData(in.getData());
+ final WrappedFileRange wfr = createFileRange(
+ range.getOffset(), range.getLength(), range.getReference());
return wfr.getInstance();
}
@@ -124,10 +141,15 @@ public final class FileRangeBridge {
* API to interact with these.
*/
private class WrappedFileRange implements VectorFileRange {
- final Object instance;
+
+ /**
+ * The wrapped range.
+ */
+ private final Object instance;
/**
* Instantiate.
+ *
* @param instance non null instance.
*/
private WrappedFileRange(final Object instance) {
@@ -162,6 +184,7 @@ public final class FileRangeBridge {
/**
* Get the instance.
+ *
* @return the instance.
*/
public Object getInstance() {
diff --git
a/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FileSystemShimImpl.java
b/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FileSystemShimImpl.java
index f00b40d..cab57f4 100644
---
a/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FileSystemShimImpl.java
+++
b/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/FileSystemShimImpl.java
@@ -118,11 +118,10 @@ public class FileSystemShimImpl extends
AbstractAPIShim<FileSystem>
// the simpler methods.
Class<FileSystem> clazz = getClazz();
- hasPathCapabilityMethod = loadInvocation(clazz, "hasPathCapability",
- Boolean.class,
+ hasPathCapabilityMethod = loadInvocation(clazz, Boolean.class,
"hasPathCapability",
Path.class, String.class);
- msyncMethod = loadInvocation(clazz, "msync", Void.class);
+ msyncMethod = loadInvocation(clazz, Void.class, "msync");
executeOpenFile = new OpenFileThroughAvailableOperation();
}
diff --git
a/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/ShimReflectionSupport.java
b/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/ShimReflectionSupport.java
index 1c4a196..804dfa6 100644
---
a/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/ShimReflectionSupport.java
+++
b/hadoop-api-shim/src/main/java/org/apache/hadoop/fs/shim/impl/ShimReflectionSupport.java
@@ -121,21 +121,23 @@ public final class ShimReflectionSupport {
*/
public static <T> Invocation<T> getInvocation(
Class<?> source, String name, Class<?>... parameterTypes) {
- return (Invocation<T>) loadInvocation(source, name, null, parameterTypes);
+ return (Invocation<T>) loadInvocation(source, null, name, parameterTypes);
}
/**
* Get an invocation from the source class, which will be unavailable() if
* the class is null or the method isn't found.
- * @param source source
- * @param name method name
+ *
* @param <T> return type
+ * @param source source
* @param returnType return type class for the compiler to be happy
+ * @param name method name
* @param parameterTypes parameters
+ *
* @return the method or "unavailable"
*/
public static <T> Invocation<T> loadInvocation(
- Class<?> source, String name, Class<? extends T> returnType, Class<?>...
parameterTypes) {
+ Class<?> source, Class<? extends T> returnType, String name, Class<?>...
parameterTypes) {
if (source == null) {
return unavailable(name);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org
For additional commands, e-mail: common-commits-h...@hadoop.apache.org
