JingsongLi commented on code in PR #10723:
URL: https://github.com/apache/paimon/pull/10723#discussion_r4345134243
##########
paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsReader.java:
##########
@@ -21,195 +21,425 @@
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.arrow.reader.ArrowBatchReader;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.columnar.AllNullColumnVector;
+import org.apache.paimon.data.columnar.ColumnVector;
+import org.apache.paimon.data.columnar.ColumnarRow;
+import org.apache.paimon.data.columnar.VectorizedColumnBatch;
+import org.apache.paimon.data.columnar.VectorizedRowIterator;
import org.apache.paimon.fs.Path;
import org.apache.paimon.reader.FileRecordIterator;
import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.ExecutorThreadFactory;
import org.apache.paimon.utils.LongIterator;
-import org.apache.paimon.utils.ProjectedRow;
+import org.apache.paimon.utils.RoaringBitmap32;
import dev.vortex.api.DataSource;
import dev.vortex.api.Expression;
import dev.vortex.api.ImmutableScanOptions;
import dev.vortex.api.Partition;
import dev.vortex.api.Scan;
import dev.vortex.api.ScanOptions;
-import dev.vortex.api.Session;
import dev.vortex.arrow.ArrowAllocation;
+import dev.vortex.arrow.ArrowBatch;
+import dev.vortex.arrow.ArrowBatchStream;
+import dev.vortex.arrow.ViewColumn;
import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.UInt8Vector;
import org.apache.arrow.vector.VectorSchemaRoot;
-import org.apache.arrow.vector.ipc.ArrowReader;
import javax.annotation.Nullable;
+import java.io.Closeable;
import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
-import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
-/** File reader for Vortex format. */
+/**
+ * File reader for Vortex format.
+ *
+ * <p>Each Arrow batch from the native scan is a new root that this reader
owns until the consumer
+ * releases the batch, so a later batch never overwrites one that the consumer
still holds. Batches
+ * are returned as {@link VectorizedRowIterator}s, so engines can consume them
column by column.
+ *
+ * <p>A pushed-down filter or a row selection drops rows in native code. The
file position of each
+ * returned row then comes from the Vortex row index, because deletion vectors
and row tracking
+ * depend on it.
+ */
public class VortexRecordsReader implements FileRecordReader<InternalRow> {
- private final ArrowBatchReader arrowBatchReader;
+ private static final String ROW_INDEX_FIELD = "_PAIMON_VORTEX_ROW_INDEX";
+
private final Path filePath;
private final BufferAllocator allocator;
- private final Session session;
+ @Nullable private final Closeable readable;
private final DataSource dataSource;
private final Scan scan;
- private final LongIterator positionIterator;
+ private final ArrowBatchReader arrowBatchReader;
@Nullable private final int[] physicalFieldMapping;
- private ArrowReader currentArrowReader;
- private Partition currentPartition;
- private long returnedPosition = -1;
-
- public VortexRecordsReader(
- Path path,
+ private final boolean readsRowIndex;
+ private final Set<ArrowBatch> activeBatches = new HashSet<>();
+
+ /** Fetches the next batch while the consumer processes the current one. */
+ private static final ExecutorService PREFETCH_POOL =
+ Executors.newCachedThreadPool(new
ExecutorThreadFactory("vortex-batch-prefetch"));
+
+ // Only one thread at a time calls fetchNext: the caller, or the prefetch
it waits for.
+ @Nullable private ArrowBatchStream currentStream;
+ @Nullable private Future<ArrowBatch> prefetched;
+ private long nextPosition;
+ private boolean closed;
+
+ VortexRecordsReader(
+ VortexInput input,
+ Path filePath,
RowType dataSchemaRowType,
RowType projectedRowType,
- @Nullable long[] rowIndices,
- @Nullable Expression predicate,
- Map<String, String> storageOptions) {
- this.filePath = path;
+ @Nullable RoaringBitmap32 selection,
+ @Nullable Expression filter) {
+ this.filePath = filePath;
+
RowType physicalReadRowType = physicalReadRowType(dataSchemaRowType,
projectedRowType);
this.physicalFieldMapping =
physicalFieldMapping(dataSchemaRowType, physicalReadRowType,
projectedRowType);
+ this.arrowBatchReader =
+ new ArrowBatchReader(physicalReadRowType, true,
VortexArrowConverters.INSTANCE);
+
+ List<String> columns = physicalReadRowType.getFieldNames();
+ this.readsRowIndex = filter != null || selection != null ||
columns.isEmpty();
+
this.allocator =
ArrowAllocation.rootAllocator()
.newChildAllocator("vortex-reader", 0, Long.MAX_VALUE);
+ Closeable openedReadable = null;
+ DataSource openedDataSource = null;
try {
- this.session = Session.create();
- try {
- this.dataSource = DataSource.open(session,
path.toUri().toString(), storageOptions);
- try {
- ImmutableScanOptions.Builder scanBuilder =
- ImmutableScanOptions.builder().ordered(true);
-
- java.util.List<String> columns =
physicalReadRowType.getFieldNames();
- // The scan clones the projection it is given, so both the
root and the select
- // built here are ours to release once dataSource.scan has
returned.
- try (Expression root = Expression.root();
- Expression projection =
- Expression.select(columns.toArray(new
String[0]), root)) {
- scanBuilder.projection(projection);
-
- if (rowIndices != null) {
- scanBuilder.selectionIndices(rowIndices);
-
scanBuilder.selectionMode(ScanOptions.SelectionMode.INCLUDE);
- }
- if (predicate != null) {
- scanBuilder.filter(predicate);
- }
- this.scan = dataSource.scan(scanBuilder.build());
- }
- } catch (Exception e) {
- dataSource.close();
- throw e;
- }
- } catch (Exception e) {
- session.close();
- throw e;
+ openedReadable = input.readable();
+ openedDataSource = input.open();
+ // The scan clones the expressions it is given, so the projection
is ours to release
+ // once the scan exists.
+ try (Expression projection = projection(columns)) {
+ this.scan = openedDataSource.scan(scanOptions(projection,
selection, filter));
}
- } catch (Exception e) {
- allocator.close();
- throw new RuntimeException(e);
+ } catch (Throwable t) {
+ closeQuietly(openedDataSource, t);
+ closeQuietly(openedReadable, t);
+ closeQuietly(allocator, t);
+ throw t;
}
- this.arrowBatchReader = new ArrowBatchReader(physicalReadRowType,
true);
- long totalRowCount = dataSource.rowCount();
- this.positionIterator =
- rowIndices != null
- ? LongIterator.fromArray(rowIndices)
- : LongIterator.fromRange(0, totalRowCount > 0 ?
totalRowCount : 0);
+ this.readable = openedReadable;
+ this.dataSource = openedDataSource;
+
+ // Start decoding the first batch now. A merge opens all files of a
split before reading
+ // any of them, so their first batches then decode in parallel.
+ this.prefetched = PREFETCH_POOL.submit(this::fetchNext);
}
@Nullable
@Override
public FileRecordIterator<InternalRow> readBatch() throws IOException {
- // Try to load the next batch from the current ArrowReader
- if (currentArrowReader != null && currentArrowReader.loadNextBatch()) {
- return toBatchIterator(currentArrowReader.getVectorSchemaRoot());
+ while (true) {
+ ArrowBatch batch = takeNext();
+ if (batch == null) {
+ return null;
+ }
+
+ track(batch);
+ if (batch.rowCount() > 0) {
+ prefetched = PREFETCH_POOL.submit(this::fetchNext);
+ return toBatchIterator(batch);
+ }
+ releaseBatch(batch);
}
+ }
- // Close current reader and move to the next partition
- closeCurrentArrowReader();
+ @Nullable
+ private ArrowBatch takeNext() throws IOException {
+ if (prefetched == null) {
+ return fetchNext();
+ }
+
+ Future<ArrowBatch> future = prefetched;
+ prefetched = null;
Review Comment:
[P1] Retain and drain the in-flight prefetch before releasing native
resources
If the reading thread is interrupted while waiting on a slow prefetch,
`await` throws InterruptedIOException with the interrupt flag restored, but
this assignment has already detached the still-running future. A subsequent
`close()` therefore skips waiting and closes the Arrow stream, scan, readable
and allocator while `fetchNext()` may still be inside native getNext/IO.
Directly interrupting close has the same problem: its interruptible `await`
exits and cleanup continues. A latch-controlled harness against these branch
classes demonstrates `readableClosedWhilePrefetchRunning=true` in both paths.
Keep the future reachable until actual completion and drain it uninterruptibly
during cleanup, restoring interruption afterward; merely cancelling the Future
is insufficient if native IO continues running.
--
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]
