JingsongLi commented on code in PR #10723:
URL: https://github.com/apache/paimon/pull/10723#discussion_r4374271738
##########
paimon-vortex/paimon-vortex-jni/pom.xml:
##########
@@ -33,7 +33,7 @@ under the License.
<properties>
<target.java.version>1.8</target.java.version>
- <vortex.jni.version>0.73.0</vortex.jni.version>
+ <vortex.jni.version>0.89.0</vortex.jni.version>
Review Comment:
[P1] Define reader upgrade and rollback protection before new Vortex writes
New evidence extends the existing Python compatibility discussion to Java:
files written through this head's actual VortexFileFormat writer for INT,
BIGINT, BOOLEAN and DOUBLE all fail in the exact baseline Java/native0.73
reader with "Encoding not found in registry: 1". A baseline0.73 writer/reader
roundtrip passes, and0.89 reads its old file. Release-2.1 also pins0.73, so a
mixed-version JVM deployment can lose reads immediately when an upgraded writer
commits even ordinary primitive rows. Please preserve compatible write
encodings, or establish and test an explicit all-readers-before-new-writes
activation contract and document that old-reader rollback after these files
exist requires rewriting/restoring them. The current changed documentation has
neither this upgrade order nor the rollback restriction.
##########
paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/arrow/ArrowBatchStream.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * 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 dev.vortex.arrow;
+
+import org.apache.arrow.c.ArrowArray;
+import org.apache.arrow.c.ArrowArrayStream;
+import org.apache.arrow.c.ArrowFormats;
+import org.apache.arrow.c.ArrowSchema;
+import org.apache.arrow.c.Data;
+import org.apache.arrow.c.NativeUtil;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.MemoryUtil;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.BitVectorHelper;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.complex.StructVector;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Reads the record batches of an Arrow C stream from the native scan. Each
batch is a new root that
+ * the caller owns and must close.
+ *
+ * <p>Vortex exports strings and binaries as {@code Utf8View} and {@code
BinaryView}. Arrow Java 15
+ * cannot import view types, and newer Arrow Java versions need Java 11. So
this stream imports view
+ * columns as {@code Utf8} and {@code Binary}: it copies their values into new
vectors. All other
+ * columns are moved into Java without a copy.
+ */
+public final class ArrowBatchStream implements AutoCloseable {
+
+ private static final long FLAG_NULLABLE = 2;
+ private static final int VIEW_SIZE = 16;
+ private static final int MAX_INLINE_VIEW_LENGTH = 12;
+
+ private final BufferAllocator allocator;
+ private final ArrowArrayStream stream;
+ private final Schema schema;
+ private final Layout layout;
+ // The schema of batch roots: all fields except the top-level view
columns, which batches
+ // expose as ViewColumns. Each child maps to its root vector, or -1 for a
view column.
+ private final Schema rootSchema;
+ private final int[] rootIndexes;
+
+ private ArrowBatchStream(
+ BufferAllocator allocator, ArrowArrayStream stream, Schema schema,
Layout layout) {
+ this.allocator = allocator;
+ this.stream = stream;
+ this.schema = schema;
+ this.layout = layout;
+
+ List<Field> rootFields = new ArrayList<>();
+ this.rootIndexes = new int[layout.children.length];
+ for (int i = 0; i < rootIndexes.length; i++) {
+ if (layout.children[i].kind == Kind.VIEW) {
+ rootIndexes[i] = -1;
+ } else {
+ rootIndexes[i] = rootFields.size();
+ rootFields.add(schema.getFields().get(i));
+ }
+ }
+ this.rootSchema = new Schema(rootFields);
+ }
+
+ /** Takes ownership of {@code stream}, which must be a stream of struct
arrays. */
+ public static ArrowBatchStream open(BufferAllocator allocator,
ArrowArrayStream stream)
+ throws IOException {
+ try (ArrowSchema cSchema = ArrowSchema.allocateNew(allocator)) {
+ stream.getSchema(cSchema);
+ try {
+ long address = cSchema.memoryAddress();
+ Field root = field(address);
+ return new ArrowBatchStream(
+ allocator, stream, new Schema(root.getChildren()),
Layout.of(address));
+ } finally {
+ cSchema.release();
+ }
+ } catch (Throwable t) {
+ stream.release();
+ stream.close();
+ throw t;
+ }
+ }
+
+ /** The schema of every batch, with view types replaced by {@code Utf8}
and {@code Binary}. */
+ public Schema schema() {
+ return schema;
+ }
+
+ /** Returns the next batch, or null at the end of the stream. */
+ public ArrowBatch next() throws IOException {
+ ArrowArray array = ArrowArray.allocateNew(allocator);
+ boolean owned = false;
+ try {
+ stream.getNext(array);
+ ArrowArray.Snapshot snapshot = array.snapshot();
+ if (snapshot.release == 0) {
+ return null;
+ }
+
+ VectorSchemaRoot root =
+ VectorSchemaRoot.create(layout.hasView ? rootSchema :
schema, allocator);
+ try {
+ if (!layout.hasView) {
+ // Moves the whole batch, then releases and closes the
array struct.
+ owned = true;
+ Data.importIntoVectorSchemaRoot(allocator, array, root,
null);
+ return new ArrowBatch(root, new ViewColumn[0], null,
root.getRowCount());
+ }
+
+ checkZeroOffset(snapshot);
+ long[] children =
+ NativeUtil.toJavaArray(snapshot.children, (int)
snapshot.n_children);
+ ViewColumn[] viewColumns = new ViewColumn[children.length];
+ for (int i = 0; i < children.length; i++) {
+ if (rootIndexes[i] < 0) {
+ // Read lazily from native memory, which the batch
keeps alive.
+ viewColumns[i] =
viewColumn(ArrowArray.wrap(children[i]).snapshot());
+ } else {
+ importColumn(
+ children[i], layout.children[i],
root.getVector(rootIndexes[i]));
+ }
+ }
+ root.setRowCount((int) snapshot.length);
+ owned = true;
+ return new ArrowBatch(root, viewColumns, array, (int)
snapshot.length);
+ } catch (Throwable t) {
+ root.close();
+ throw t;
+ }
+ } finally {
+ if (!owned) {
+ if (array.snapshot().release != 0) {
+ array.release();
+ }
+ array.close();
+ }
+ }
+ }
+
+ private static ViewColumn viewColumn(ArrowArray.Snapshot snapshot) {
+ long[] buffers = NativeUtil.toJavaArray(snapshot.buffers, (int)
snapshot.n_buffers);
+ // Buffers: validity, views, the data buffers, then the variadic
buffer sizes.
+ long[] dataBuffers = new long[buffers.length - 3];
+ System.arraycopy(buffers, 2, dataBuffers, 0, dataBuffers.length);
+ return new ViewColumn(
+ buffers[0], buffers[1], dataBuffers, snapshot.offset, (int)
snapshot.length);
+ }
+
+ @Override
+ public void close() {
+ if (stream.snapshot().release != 0) {
+ stream.release();
+ }
+ stream.close();
+ }
+
+ private void importColumn(long arrayAddress, Layout layout, FieldVector
vector) {
+ if (!layout.hasView) {
+ // Moves the column, and marks it as released so its parent does
not release it.
+ Data.importIntoVector(allocator, ArrowArray.wrap(arrayAddress),
vector, null);
+ return;
+ }
+
+ ArrowArray.Snapshot snapshot =
ArrowArray.wrap(arrayAddress).snapshot();
+ switch (layout.kind) {
+ case VIEW:
+ copyViews(snapshot, (BaseVariableWidthVector) vector);
+ return;
+ case STRUCT:
+ importStruct(snapshot, layout, (StructVector) vector);
+ return;
+ case LIST:
+ importList(snapshot, layout, (ListVector) vector);
+ return;
+ default:
+ throw new UnsupportedOperationException(
+ "Cannot import a view type nested in " +
vector.getField());
+ }
+ }
+
+ private void importStruct(ArrowArray.Snapshot snapshot, Layout layout,
StructVector vector) {
+ checkZeroOffset(snapshot);
+ int length = (int) snapshot.length;
+ long validity = buffer(snapshot, 0);
+
+ long[] children = NativeUtil.toJavaArray(snapshot.children, (int)
snapshot.n_children);
+ for (int i = 0; i < children.length; i++) {
+ importColumn(
+ children[i], layout.children[i], (FieldVector)
vector.getChildByOrdinal(i));
+ }
+
+ for (int i = 0; i < length; i++) {
+ if (isValid(validity, i)) {
+ vector.setIndexDefined(i);
+ } else {
+ vector.setNull(i);
+ }
+ }
+ vector.setValueCount(length);
+ }
+
+ private void importList(ArrowArray.Snapshot snapshot, Layout layout,
ListVector vector) {
+ checkZeroOffset(snapshot);
+ int length = (int) snapshot.length;
+ long validity = buffer(snapshot, 0);
+ long offsets = buffer(snapshot, 1);
+ if (length > 0 && MemoryUtil.UNSAFE.getInt(offsets) != 0) {
+ throw new UnsupportedOperationException(
+ "Cannot import a list whose offsets do not start at 0");
Review Comment:
[P1] Import valid sliced list offsets instead of rejecting filtered arrays
A native pushed filter can legitimately return a list whose first child
offset is nonzero. With a real 100-row file containing id INT, arr
ARRAY<STRING> and a nested ROW, the unfiltered read succeeds, but id >= 80
throws this UnsupportedOperationException instead of returning20 rows.
Selection plus filtering fails too; selection alone succeeds. The exact
baseline format/JNI sources with native0.73 read all four modes correctly.
Normalize/rebase the list offsets and child storage consistently (including
null lists), and add a filtered nested-string regression; ordinary filtered
scans of these supported schemas must remain readable.
##########
paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsReader.java:
##########
@@ -21,195 +21,460 @@
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();
+ }
- while (scan.hasNext()) {
- closeCurrentPartition();
- currentPartition = scan.next();
- currentArrowReader = currentPartition.scanArrow(allocator);
- if (currentArrowReader.loadNextBatch()) {
- return
toBatchIterator(currentArrowReader.getVectorSchemaRoot());
+ // Keep the prefetch reachable until it completes. An interrupted wait
leaves it running,
+ // and close must then wait for it before it releases native resources.
+ try {
+ return await(prefetched);
+ } finally {
+ if (prefetched.isDone()) {
Review Comment:
[P2] Retain an interrupted prefetch even if it just completed
Future.get can throw InterruptedException and the task can finish before
this finally block observes isDone. Clearing prefetched then loses the returned
ArrowBatch: readBatch never calls track, and close cannot retrieve or release
it. A latch-controlled real FutureTask with an allocated Arrow vector
reproduces InterruptedIOException, prefetchRetained=false and reader.close
throwing "Memory was leaked" (16 bytes in the minimal fixture); the same
interleaving releases everything when the future is retained after
interruption. Clear the future after actually consuming its result (or a
terminal failure), while keeping it reachable after an interrupted wait even
when it has just completed.
--
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]
