JingsongLi commented on code in PR #10723: URL: https://github.com/apache/paimon/pull/10723#discussion_r4374271733
########## 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 returning 20 rows. Selection plus filtering fails too; selection alone succeeds. The exact baseline format/JNI sources with native 0.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-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/native 0.73 reader with "Encoding not found in registry: 1". A baseline 0.73 writer/reader roundtrip passes, and 0.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. -- 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]
