steveloughran commented on code in PR #1139:
URL: https://github.com/apache/parquet-mr/pull/1139#discussion_r1552294730
##########
pom.xml:
##########
@@ -590,6 +590,8 @@
<exclude>org.apache.parquet.conf.PlainParquetConfiguration#getClass(java.lang.String,java.lang.Class,java.lang.Class)</exclude>
<exclude>org.apache.parquet.conf.ParquetConfiguration#getClass(java.lang.String,java.lang.Class,java.lang.Class)</exclude>
<exclude>org.apache.parquet.hadoop.util.SerializationUtil#readObjectFromConfAsBase64(java.lang.String,org.apache.parquet.conf.ParquetConfiguration)</exclude>
+
<exclude>org.apache.parquet.hadoop.util.wrappedio.FutureIO#awaitFuture(java.util.concurrent.Future,long,java.util.concurrent.TimeUnit)</exclude>
Review Comment:
Oh, it isn't: but if I don't do this the checker plugin fails complaining
that it *is*. If there is another way to get it to leave it alone -happy to do
that instead.
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/wrappedio/BindingUtils.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.parquet.hadoop.util.wrappedio;
+
+import org.apache.parquet.util.DynMethods;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utility methods to assist binding to Hadoop APIs through reflection.
+ */
+public final class BindingUtils {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(BindingUtils.class);
+
+ private BindingUtils() {}
+
+ /**
+ * Get an invocation from the source class, which will be unavailable() if
+ * the class is null or the method isn't found.
+ *
+ * @param <T> return type
+ * @param source source. If null, the method is a no-op.
+ * @param returnType return type class (unused)
+ * @param name method name
+ * @param parameterTypes parameters
+ *
+ * @return the method or "unavailable"
+ */
+ static <T> DynMethods.UnboundMethod loadInvocation(
+ Class<?> source, Class<? extends T> returnType, String name, Class<?>...
parameterTypes) {
+
+ if (source != null) {
+ final DynMethods.UnboundMethod m = new DynMethods.Builder(name)
+ .impl(source, name, parameterTypes)
+ .orNoop()
+ .build();
+ if (m.isNoop()) {
+ // this is a sign of a mismatch between this class's expected
+ // signatures and actual ones.
+ // log at debug.
+ LOG.debug("Failed to load method {} from {}", name, source);
+ } else {
+ LOG.debug("Found method {} from {}", name, source);
+ }
+ return m;
+ } else {
+ return noop(name);
+ }
+ }
+
+ /**
+ * Create a no-op method.
+ *
+ * @param name method name
+ *
+ * @return a no-op method.
+ */
+ static DynMethods.UnboundMethod noop(final String name) {
Review Comment:
i'm not giving an implementation class, just a method name which is required
for build()...it is always null if my reading of the code is therefore always
null
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/wrappedio/FutureIO.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.parquet.hadoop.util.wrappedio;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.io.UncheckedIOException;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Methods to work with futures, based on.
+ * {@code org.apache.hadoop.util.functional.FutureIO}.
+ *
+ * These methods are used in production code.
+ */
+public final class FutureIO {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FutureIO.class);
+
+ /**
+ * Given a future, evaluate it.
+ * <p>
+ * Any exception generated in the future is
+ * extracted and rethrown.
+ * </p>
+ *
+ * @param future future to evaluate
+ * @param timeout timeout to wait
+ * @param unit time unit.
+ * @param <T> type of the result.
+ *
+ * @return the result, if all went well.
+ *
+ * @throws InterruptedIOException future was interrupted
+ * @throws IOException if something went wrong
+ * @throws RuntimeException any nested RTE thrown
+ * @throws TimeoutException the future timed out.
+ */
+ public static <T> T awaitFuture(final Future<T> future, final long timeout,
final TimeUnit unit)
+ throws InterruptedIOException, IOException, RuntimeException,
TimeoutException {
+ try {
+ LOG.debug("Awaiting future");
+ return future.get(timeout, unit);
+ } catch (InterruptedException e) {
+ throw (InterruptedIOException) new
InterruptedIOException(e.toString()).initCause(e);
+ } catch (ExecutionException e) {
Review Comment:
valid point; makes sense. this also means that the hadoop FutureIO classes
should do the same.
looking into the code, we are picking that up in
unwrapInnerException(), which triggers an extract-and-recurse loop, but it's
not being triggered if the base exception raised is CompletionException. changed
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/wrappedio/VectorIOBridge.java:
##########
@@ -0,0 +1,422 @@
+/*
+ * 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.parquet.hadoop.util.wrappedio;
+
+import static java.util.Objects.requireNonNull;
+import static org.apache.parquet.Exceptions.throwIfInstance;
+import static org.apache.parquet.Preconditions.checkArgument;
+import static
org.apache.parquet.hadoop.util.wrappedio.BindingUtils.loadInvocation;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.PositionedReadable;
+import org.apache.parquet.bytes.ByteBufferAllocator;
+import org.apache.parquet.io.ParquetFileRange;
+import org.apache.parquet.util.DynMethods;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Vector IO bridge.
+ * <p>
+ * This loads the {@code PositionedReadable} method:
+ * <pre>
+ * void readVectored(List[?extends FileRange] ranges,
+ * IntFunction[ByteBuffer] allocate) throws IOException
+ * </pre>
+ * It is made slightly easier because of type erasure; the signature of the
+ * function is actually {@code Void.class method(List.class,
IntFunction.class)}.
+ * <p>
+ * There are some counters to aid in testing; the {@link #toString()} method
+ * will print them and the loaded method, for use in tests and debug logs.
+ */
+public final class VectorIOBridge {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(VectorIOBridge.class);
+
+ /**
+ * readVectored Method to look for.
+ */
+ private static final String READ_VECTORED = "readVectored";
+
+ /**
+ * hasCapability Method name.
+ * {@code boolean hasCapability(String capability);}
+ */
+ private static final String HAS_CAPABILITY = "hasCapability";
+
+ /**
+ * hasCapability() probe for vectored IO api implementation.
+ */
+ static final String VECTOREDIO_CAPABILITY = "in:readvectored";
+
+ /**
+ * The singleton instance of the bridge.
+ */
+ private static final VectorIOBridge INSTANCE = new VectorIOBridge();
+
+ /**
+ * readVectored() method.
+ */
+ private final DynMethods.UnboundMethod readVectored;
+
+ /**
+ * {@code boolean StreamCapabilities.hasCapability(String)}.
+ */
+ private final DynMethods.UnboundMethod hasCapabilityMethod;
+
+ /**
+ * How many vector read calls made.
+ */
+ private final AtomicLong vectorReads = new AtomicLong();
+
+ /**
+ * How many blocks were read.
+ */
+ private final AtomicLong blocksRead = new AtomicLong();
+
+ /**
+ * How many bytes were read.
+ */
+ private final AtomicLong bytesRead = new AtomicLong();
+
+ /**
+ * Constructor. package private for testing.
+ */
+ private VectorIOBridge() {
+
+ readVectored =
+ loadInvocation(PositionedReadable.class, Void.TYPE, READ_VECTORED,
List.class, IntFunction.class);
+ LOG.debug("Vector IO availability: {}", available());
+
+ // if readVectored is present, so is hasCapabilities().
+ hasCapabilityMethod = loadInvocation(FSDataInputStream.class,
boolean.class, HAS_CAPABILITY, String.class);
+ }
+
+ /**
+ * Is the vectored IO API available for the given stream
+ * and allocator in this Hadoop runtime?
+ *
+ * @param stream input stream to query.
+ * @param allocator allocator to be used.
+ *
+ * @return true if the stream declares the capability is available.
+ */
+ public boolean readVectoredAvailable(final FSDataInputStream stream, final
ByteBufferAllocator allocator) {
+ return available() && !allocator.isDirect();
+ }
+
+ /**
+ * Is the bridge available?
+ *
+ * @return true if readVectored() is available.
+ */
+ public boolean available() {
+ return !readVectored.isNoop() && FileRangeBridge.bridgeAvailable();
+ }
+
+ /**
+ * Check that the bridge is available.
+ *
+ * @throws UnsupportedOperationException if it is not.
+ */
+ private void checkAvailable() {
+ if (!available()) {
+ throw new UnsupportedOperationException("Hadoop VectorIO not found");
+ }
+ }
+
+ /**
+ * Read fully a list of file ranges asynchronously from this file.
+ * The default iterates through the ranges to read each synchronously, but
+ * the intent is that FSDataInputStream subclasses can make more efficient
+ * readers.
+ * The {@link ParquetFileRange} parameters all have their
+ * data read futures set to the range reads of the associated
+ * operations; callers must await these to complete.
+ * <p>
+ * As a result of the call, each range will have
FileRange.setData(CompletableFuture)
+ * called with a future that when complete will have a ByteBuffer with the
+ * data from the file's range.
+ * <p>
+ * The position returned by getPos() after readVectored() is undefined.
+ * </p>
+ * <p>
+ * If a file is changed while the readVectored() operation is in progress,
the output is
+ * undefined. Some ranges may have old data, some may have new and some
may have both.
+ * </p>
+ * <p>
+ * While a readVectored() operation is in progress, normal read api calls
may block.
+ * </p>
+ * @param stream stream from where the data has to be read.
+ * @param ranges parquet file ranges.
+ * @param allocator buffer allocator.
+ * @throws UnsupportedOperationException if the API is not available.
+ * @throws EOFException if a range is past the end of the file.
+ * @throws IOException other IO problem initiating the read operations.
+ */
+ public static void readVectoredRanges(
Review Comment:
will do
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1171,6 +1176,108 @@ public ColumnChunkPageReadStore
readFilteredRowGroup(int blockIndex, RowRanges r
return internalReadFilteredRowGroup(block, rowRanges,
getColumnIndexStore(blockIndex));
}
+ /**
+ * Read data in all parts via either vectored IO or serial IO.
+ * @param allParts all parts to be read.
+ * @param builder used to build chunk list to read the pages for the
different columns.
+ * @throws IOException any IOE.
+ */
+ private void readAllPartsVectoredOrNormal(List<ConsecutivePartList>
allParts, ChunkListBuilder builder)
+ throws IOException {
+
+ if (shouldUseVectoredIO(allParts)) {
+ try {
+ readVectored(allParts, builder);
+ return;
+ } catch (IllegalArgumentException | UnsupportedOperationException e) {
+ // Either the arguments are wrong or somehow this is being invoked
against
+ // a hadoop release which doesn't have the API and yet somehow it got
here.
+ LOG.warn("readVectored() failed; falling back to normal IO against
{}", f, e);
+ }
+ }
+ for (ConsecutivePartList consecutiveChunks : allParts) {
+ consecutiveChunks.readAll(f, builder);
+ }
+ }
+
+ /**
+ * Should the read use vectored IO?
+ * <p>
+ * This returns true if all necessary conditions are met:
+ * <ol>
+ * <li> The option is enabled</li>
+ * <li> The Hadoop version supports vectored IO</li>
+ * <li> The part lengths are all valid for vectored IO</li>
+ * <li> The stream implementation explicitly supports the API; for other
streams the classic
+ * API is always used.</li>
+ * <li> The allocator is not direct. This is to avoid HADOOP-19101
surfacing.
+ * </ol>
+ * @param allParts all parts to read.
+ * @return true or false.
+ */
+ private boolean shouldUseVectoredIO(final List<ConsecutivePartList>
allParts) {
+ return options.useHadoopVectoredIO()
+ && f.readVectoredAvailable(options.getAllocator())
+ && arePartsValidForVectoredIO(allParts);
+ }
+
+ /**
+ * Validated the parts for vectored IO.
+ * Vectored IO doesn't support reading ranges of size greater than
+ * Integer.MAX_VALUE.
+ * @param allParts all parts to read.
+ * @return true or false.
+ */
+ private boolean arePartsValidForVectoredIO(List<ConsecutivePartList>
allParts) {
+ for (ConsecutivePartList consecutivePart : allParts) {
+ if (consecutivePart.length >= Integer.MAX_VALUE) {
+ LOG.debug(
+ "Part length {} greater than Integer.MAX_VALUE thus disabling
vectored IO",
+ consecutivePart.length);
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Read all parts through vectored IO.
+ * <p>
+ * The API is available in recent hadoop builds for all implementations of
PositionedReadable;
+ * the default implementation simply does a sequence of reads at different
offsets.
+ * <p>
+ * If directly implemented by a Filesystem then it is likely to be a more
efficient
+ * operation such as a scatter-gather read (native IO) or set of parallel
+ * GET requests against an object store.
+ * @param allParts all parts to be read.
+ * @param builder used to build chunk list to read the pages for the
different columns.
+ * @throws IOException any IOE.
+ * @throws IllegalArgumentException arguments are invalid.
+ * @throws UnsupportedOperationException if the filesystem does not support
vectored IO.
+ */
+ private void readVectored(List<ConsecutivePartList> allParts,
ChunkListBuilder builder) throws IOException {
+
+ List<ParquetFileRange> ranges = new ArrayList<>(allParts.size());
+ long totalSize = 0;
+ for (ConsecutivePartList consecutiveChunks : allParts) {
+ final long len = consecutiveChunks.length;
+ Preconditions.checkArgument(
Review Comment:
just for strictness and testing...happy to cut though it's a low cost
operation compared to the reading.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
