SemyonSinchenko commented on code in PR #975: URL: https://github.com/apache/incubator-graphar/pull/975#discussion_r4027380491
########## maven-projects/io-api/src/main/java/org/apache/graphar/io/RecordBatches.java: ########## @@ -0,0 +1,81 @@ +/* + * 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.graphar.io; + +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; + +/** Builds {@link RecordBatch} instances from values a producer holds row by row. */ +public final class RecordBatches { + private RecordBatches() {} + + /** + * Transposes row-major values into a columnar batch. + * + * @param schema the batch schema; every row must hold one value per field + * @param rows the rows, each a list of boxed values in schema order + * @return a batch backed by {@link ObjectValueVector} columns + */ + public static RecordBatch ofRows(Schema schema, List<? extends List<?>> rows) { Review Comment: `ofRows` validates only the row width; neither value types nor nullability are checked against the Field definitions. A String in an INT64 column, or a null in a non-nullable column, passes construction silently and only fails later deep inside a format encoder (e.g., `ParquetPhysicalWriter` 's "Unexpected value for ..." / "Required field is null"), far from the site where the bad row was supplied. Since `Field` exposes `type()` and `nullable()`, failing fast here — at minimum rejecting nulls in non-nullable columns — would make producer errors immediately attributable. If deferred validation is intentional, document it on the factory. ########## maven-projects/io-api/src/main/java/org/apache/graphar/io/ObjectValueVector.java: ########## @@ -0,0 +1,66 @@ +/* + * 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.graphar.io; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +/** + * A {@link ValueVector} over boxed values held in memory. + * + * <p>This is the vector a producer uses when it assembles a batch from Java objects rather than + * decoding one from a file. Values are snapshotted at construction and list values are held as + * unmodifiable copies, so neither the producer nor a consumer can mutate the vector afterwards. + */ +public final class ObjectValueVector implements ValueVector { + private final Field field; + private final List<Object> values; + + public ObjectValueVector(Field field, List<?> values) { + this.field = Objects.requireNonNull(field, "A vector field cannot be null."); + Objects.requireNonNull(values, "Vector values cannot be null."); + List<Object> copy = new ArrayList<>(values.size()); + for (Object value : values) { Review Comment: The snapshot is only one level deep. For nested list values (`ColumnType.listOf(listOf(...)`) is a representable type), the inner lists are neither copied nor made unmodifiable, so the producer's original inner lists remain live references inside the vector and can be mutated after construction. This contradicts the class javadoc's guarantee that neither the producer nor a consumer can mutate the vector afterwards. Either snapshot list values recursively, or narrow the javadoc to state that only the top level of list values is protected. ########## maven-projects/io-api/src/main/java/org/apache/graphar/io/ObjectValueVector.java: ########## @@ -0,0 +1,66 @@ +/* + * 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.graphar.io; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +/** + * A {@link ValueVector} over boxed values held in memory. + * + * <p>This is the vector a producer uses when it assembles a batch from Java objects rather than + * decoding one from a file. Values are snapshotted at construction and list values are held as + * unmodifiable copies, so neither the producer nor a consumer can mutate the vector afterwards. + */ +public final class ObjectValueVector implements ValueVector { + private final Field field; + private final List<Object> values; + + public ObjectValueVector(Field field, List<?> values) { + this.field = Objects.requireNonNull(field, "A vector field cannot be null."); + Objects.requireNonNull(values, "Vector values cannot be null."); + List<Object> copy = new ArrayList<>(values.size()); + for (Object value : values) { + copy.add(value instanceof List ? List.copyOf((List<?>) value) : value); Review Comment: `List.copyOf` throws a message-less `NullPointerException` when a list value contains null elements, yet the io-api type model explicitly permits such values: `ColumnType.listOf(elementType)` builds a list whose element field is nullable, and `listOfElement` preserves element nullability. A producer assembling a batch for a nullable-element list column that contains a null entry therefore cannot construct this vector at all, and fails with an opaque NPE that points neither at the column nor the offending value. Use a null-tolerant unmodifiable copy (e.g., `Collections.unmodifiableList(new ArrayList<>(...)))` so the vector can represent every value its Field's type allows. ########## maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetPhysicalReader.java: ########## @@ -0,0 +1,426 @@ +/* + * 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.graphar.io.parquet; + +import java.io.IOException; +import java.net.URI; +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.graphar.io.ColumnRef; +import org.apache.graphar.io.ColumnType; +import org.apache.graphar.io.Field; +import org.apache.graphar.io.PhysicalReader; +import org.apache.graphar.io.ReadCapability; +import org.apache.graphar.io.ReadReport; +import org.apache.graphar.io.ReadRequest; +import org.apache.graphar.io.ReadResult; +import org.apache.graphar.io.Schema; +import org.apache.graphar.storage.InputFile; +import org.apache.graphar.storage.Storage; +import org.apache.parquet.ParquetReadOptions; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.io.SeekableInputStream; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Type; + +/** A storage-backed reader for GraphAr primitive and LIST Parquet fields. */ +public final class ParquetPhysicalReader implements PhysicalReader { + private static final Set<ReadCapability> CAPABILITIES = + Collections.unmodifiableSet( + EnumSet.of( + ReadCapability.PROJECTION, + ReadCapability.ROW_RANGE, + ReadCapability.LIMIT)); + + private static final int DEFAULT_FOOTER_CACHE_CAPACITY = 256; + + private final Storage storage; + private final FooterCache footers; + + /** + * Creates a reader that resolves each request URI through {@code storage} and remembers the + * footers of recently read files. + */ + public ParquetPhysicalReader(Storage storage) { + this(storage, DEFAULT_FOOTER_CACHE_CAPACITY); + } + + /** + * Creates a reader that keeps at most {@code footerCacheCapacity} Parquet footers in memory. + * GraphAr chunk files are immutable once published, so repeated range reads of one chunk parse + * its footer once. A capacity of zero parses the footer on every request. + * + * @param storage resolves each request URI to a readable file + * @param footerCacheCapacity maximum number of remembered footers + */ + public ParquetPhysicalReader(Storage storage, int footerCacheCapacity) { + this.storage = Objects.requireNonNull(storage, "storage"); + this.footers = new FooterCache(footerCacheCapacity); + } + + @Override + public Set<ReadCapability> capabilities() { + return CAPABILITIES; + } + + @Override + public ReadResult read(ReadRequest request) throws IOException { + Objects.requireNonNull(request, "request"); + if (!request.filters().isEmpty()) { + throw new UnsupportedOperationException( + "Parquet filter pushdown is not implemented; refusing semantic fallback."); + } + InputFile inputFile = + Objects.requireNonNull(storage.inputFile(request.uri()), "storage inputFile"); + ParquetFileReader fileReader = null; + try { + fileReader = open(request.uri(), new ParquetInputFile(inputFile)); + MessageType fileSchema = fileReader.getFooter().getFileMetaData().getSchema(); + List<ParquetColumn> fileColumns = columns(fileSchema); + Map<String, ParquetColumn> columnsByName = byName(fileColumns); + List<ParquetColumn> outputColumns = outputColumns(request, fileColumns, columnsByName); + List<ParquetColumn> readColumns = readColumns(fileColumns, outputColumns); + MessageType readSchema = + new MessageType(fileSchema.getName(), parquetTypes(readColumns)); + fileReader.setRequestedSchema(readSchema); + + Schema outputSchema = new Schema(fields(outputColumns)); + ReadReport report = new ReadReport(applied(request), declined(request)); + ParquetBatchCursor cursor = + new ParquetBatchCursor( + fileReader, + fileSchema, + readSchema, + readColumns, + outputColumns, + outputSchema, + request); + fileReader = null; + return new ReadResult(request, cursor, report); + } finally { + if (fileReader != null) { + fileReader.close(); + } + } + } + + /** + * Opens a Parquet reader, reusing a remembered footer when the file is unchanged in size. The + * returned reader owns the stream opened here and closes it. + */ + private ParquetFileReader open(URI uri, ParquetInputFile file) throws IOException { + ParquetReadOptions options = ParquetReadOptions.builder().build(); + long size = file.getLength(); + ParquetMetadata remembered = footers.get(uri, size); + SeekableInputStream stream = file.newStream(); + try { + ParquetMetadata footer = remembered; + if (footer == null) { + footer = ParquetFileReader.readFooter(file, options, stream); + if (isPlaintext(footer)) { + footers.put(uri, size, footer); + } + } + ParquetFileReader reader = ParquetFileReader.open(file, footer, options, stream); + stream = null; + return reader; + } finally { + if (stream != null) { + stream.close(); + } + } + } + + /** + * Returns whether a footer is safe to remember. An encrypted file carries a stateful decryptor + * inside its metadata, so its footer is parsed again for every reader. + */ + private static boolean isPlaintext(ParquetMetadata footer) { + return footer.getFileMetaData().getFileDecryptor() == null; + } + + private static List<ParquetColumn> columns(MessageType fileSchema) { + List<ParquetColumn> columns = new ArrayList<>(); + for (Type type : fileSchema.getFields()) { + if (type.isRepetition(Type.Repetition.REPEATED) + || (!type.isPrimitive() && !isList(type))) { + throw new IllegalArgumentException( + "Only primitive or standard LIST Parquet columns are supported: " + + type.getName()); + } + columns.add(new ParquetColumn(type, toField(type))); + } + return List.copyOf(columns); + } + + private static Map<String, ParquetColumn> byName(List<ParquetColumn> columns) { + Map<String, ParquetColumn> columnsByName = new HashMap<>(); + for (ParquetColumn column : columns) { + if (columnsByName.put(column.field().name(), column) != null) { + throw new IllegalArgumentException( + "Duplicate Parquet column: " + column.field().name()); + } + } + return columnsByName; + } + + private static List<ParquetColumn> outputColumns( + ReadRequest request, + List<ParquetColumn> fileColumns, + Map<String, ParquetColumn> columnsByName) { + if (request.projection().isAllColumns()) { + return fileColumns; + } + List<ParquetColumn> result = new ArrayList<>(); + for (ColumnRef reference : request.projection().columns()) { + ParquetColumn column = columnsByName.get(reference.name()); + if (column == null) { + throw new IllegalArgumentException("Unknown projection column: " + reference); + } + result.add(column); + } + return List.copyOf(result); + } + + private static List<ParquetColumn> readColumns( + List<ParquetColumn> fileColumns, List<ParquetColumn> outputColumns) { + Map<String, ParquetColumn> needed = new LinkedHashMap<>(); + for (ParquetColumn column : outputColumns) { + needed.put(column.field().name(), column); + } + List<ParquetColumn> result = new ArrayList<>(); + for (ParquetColumn column : fileColumns) { + if (needed.containsKey(column.field().name())) { + result.add(column); + } + } + return List.copyOf(result); + } + + private static List<Type> parquetTypes(List<ParquetColumn> columns) { + List<Type> types = new ArrayList<>(); + for (ParquetColumn column : columns) { + types.add(column.parquetType()); + } + return types; + } + + private static List<Field> fields(List<ParquetColumn> columns) { + List<Field> fields = new ArrayList<>(); + for (ParquetColumn column : columns) { + fields.add(column.field()); + } + return fields; + } + + private static Set<ReadCapability> applied(ReadRequest request) { + EnumSet<ReadCapability> applied = EnumSet.noneOf(ReadCapability.class); + if (!request.projection().isAllColumns()) { + applied.add(ReadCapability.PROJECTION); + } + if (request.rowRange().isPresent()) { + applied.add(ReadCapability.ROW_RANGE); + } + if (request.limit().isPresent()) { + applied.add(ReadCapability.LIMIT); + } + return applied; + } + + private static Set<ReadCapability> declined(ReadRequest request) { + return Collections.emptySet(); + } + + private static boolean isList(Type type) { + return type.getLogicalTypeAnnotation() + instanceof LogicalTypeAnnotation.ListLogicalTypeAnnotation; + } + + private static Field toField(Type type) { + return new Field(type.getName(), type(type), !type.isRepetition(Type.Repetition.REQUIRED)); + } + + private static ColumnType type(Type type) { + if (!type.isPrimitive()) { + return listType(type); + } + PrimitiveType primitive = type.asPrimitiveType(); + return primitiveType(primitive); + } + + private static ColumnType listType(Type type) { + if (!isList(type) || type.asGroupType().getFieldCount() != 1) { + throw new IllegalArgumentException("Unsupported Parquet LIST field: " + type); + } + Type repeated = type.asGroupType().getType(0); + if (!repeated.isRepetition(Type.Repetition.REPEATED) + || repeated.isPrimitive() + || repeated.asGroupType().getFieldCount() != 1) { + throw new IllegalArgumentException("Unsupported Parquet LIST field: " + type); + } + Type element = repeated.asGroupType().getType(0); + if (!element.isPrimitive() || !element.isRepetition(Type.Repetition.REPEATED)) { Review Comment: This requires the LIST element to be REPEATED, but the Parquet spec's 3-level LIST layout defines the element as REQUIRED or OPTIONAL. Files written by Spark/Arrow/pyarrow are therefore rejected here at schema-mapping time, contradicting the "Only primitive or standard LIST Parquet columns are supported" message in columns(). Note that ParquetBatchCursor.listValue already reads values through the middle repeated "list" wrapper group, so accepting spec-conformant elements mainly needs this check relaxed plus null-element handling (element repetition count 0). Also note this check and `ParquetPhysicalWriter.listElementType` must be changed together: the writer currently emits a REPEATED element, so today the module only round-trips its own non-standard layout. ########## maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetPhysicalWriter.java: ########## @@ -0,0 +1,288 @@ +/* + * 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.graphar.io.parquet; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.time.Instant; +import java.time.LocalDate; +import java.util.List; +import java.util.Objects; +import org.apache.graphar.io.BatchCursor; +import org.apache.graphar.io.ColumnType; +import org.apache.graphar.io.Field; +import org.apache.graphar.io.PhysicalWriter; +import org.apache.graphar.io.RecordBatch; +import org.apache.graphar.io.Schema; +import org.apache.graphar.io.WriteMode; +import org.apache.graphar.io.WriteRequest; +import org.apache.graphar.storage.Storage; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.ParquetFileWriter; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Type; +import org.apache.parquet.schema.Types; + +/** A storage-backed writer for flat, primitive Parquet batches. */ +public final class ParquetPhysicalWriter implements PhysicalWriter { + private static final int INDEXED_PAGE_ROW_COUNT = 1024; + + private final Storage storage; + + /** + * Maps the request mode onto a Parquet file mode. Parquet cannot append to a closed file + * without rewriting it, so {@link WriteMode#APPEND} is rejected instead of being downgraded to + * an overwrite. + */ + static ParquetFileWriter.Mode fileMode(WriteMode mode) { + switch (Objects.requireNonNull(mode, "mode")) { + case CREATE_NEW: + return ParquetFileWriter.Mode.CREATE; + case OVERWRITE: + return ParquetFileWriter.Mode.OVERWRITE; + default: + throw new UnsupportedOperationException( + "Parquet cannot append to an existing file; write mode " + mode); + } + } + + /** Creates a writer that resolves every output URI through {@code storage}. */ + public ParquetPhysicalWriter(Storage storage) { + this.storage = Objects.requireNonNull(storage, "storage"); + } + + @Override + public void write(WriteRequest request, BatchCursor batches) throws IOException { + Objects.requireNonNull(request, "request"); + Objects.requireNonNull(batches, "batches"); + MessageType parquetSchema = parquetSchema(request.schema()); + ParquetFileWriter.Mode mode = fileMode(request.mode()); + try (ParquetWriter<Group> writer = + ExampleParquetWriter.builder( + new ParquetOutputFile(storage.outputFile(request.uri()))) + .withType(parquetSchema) + .withWriteMode(mode) + .withPageRowCountLimit(INDEXED_PAGE_ROW_COUNT) + .build()) { + SimpleGroupFactory groups = new SimpleGroupFactory(parquetSchema); + while (batches.next()) { + RecordBatch batch = + Objects.requireNonNull(batches.batch(), "batch cursor returned null"); + requireSchema(request.schema(), batch.schema()); + for (int rowIndex = 0; rowIndex < batch.rowCount(); rowIndex++) { + writer.write(toGroup(groups, request.schema(), batch, rowIndex)); + } + } + } finally { + batches.close(); + } + } + + private static MessageType parquetSchema(Schema schema) { + Types.MessageTypeBuilder builder = Types.buildMessage(); + for (Field field : schema.fields()) { + if (field.type().kind() == ColumnType.Kind.LIST) { + builder.addField(listType(field)); + continue; + } + Types.PrimitiveBuilder<Types.GroupBuilder<MessageType>> primitive = + builder.primitive(physicalType(field.type()), repetition(field)); + LogicalTypeAnnotation logicalType = logicalType(field.type()); + if (logicalType != null) { + primitive.as(logicalType); + } + primitive.named(field.name()); + } + return builder.named("graphar"); + } + + private static Type listType(Field field) { + ColumnType element = field.type().elementType().orElseThrow(); + Type value = listElementType(element); + return (field.nullable() ? Types.optionalList() : Types.requiredList()) + .element(value) + .named(field.name()); + } + + private static Type listElementType(ColumnType element) { + LogicalTypeAnnotation logical = logicalType(element); + if (logical == null) { + return Types.repeated(physicalType(element)).named("element"); Review Comment: See the comment in Reader about "repeated" ########## maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetBatchCursor.java: ########## @@ -0,0 +1,350 @@ +/* + * 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.graphar.io.parquet; + +import java.io.IOException; +import java.time.Instant; +import java.time.LocalDate; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.graphar.io.BatchCursor; +import org.apache.graphar.io.ReadRequest; +import org.apache.graphar.io.RecordBatch; +import org.apache.graphar.io.RowRange; +import org.apache.graphar.io.Schema; +import org.apache.graphar.io.ValueVector; +import org.apache.graphar.io.VectorRecordBatch; +import org.apache.parquet.column.page.PageReadStore; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.convert.GroupRecordConverter; +import org.apache.parquet.filter2.columnindex.RowRanges; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.internal.filter2.columnindex.ColumnIndexStore.MissingOffsetIndexException; +import org.apache.parquet.io.ColumnIOFactory; +import org.apache.parquet.io.MessageColumnIO; +import org.apache.parquet.io.RecordReader; +import org.apache.parquet.schema.MessageType; + +/** Streams materialized Parquet row groups as neutral record batches. */ +final class ParquetBatchCursor implements BatchCursor { + private static final int BATCH_ROWS = 1_024; + private final ParquetFileReader fileReader; + private final MessageType fileSchema; + private final MessageType readSchema; + private final List<ParquetColumn> readColumns; + private final Schema outputSchema; + private final Map<String, Integer> readColumnIndexes; + private final int[] outputIndexes; + private final long rangeStart; + private final long rangeEnd; + private final long limit; + private final List<BlockRange> rowGroups; + private int nextRowGroup; + private long emitted; + private long rowsRemainingInGroup; + private PageReadStore pages; + private RecordReader<Group> rows; + private boolean exhausted; + private boolean closed; + private RecordBatch current; + + ParquetBatchCursor( + ParquetFileReader fileReader, + MessageType fileSchema, + MessageType readSchema, + List<ParquetColumn> readColumns, + List<ParquetColumn> outputColumns, + Schema outputSchema, + ReadRequest request) { + this.fileReader = fileReader; + this.fileSchema = fileSchema; + this.readSchema = readSchema; + this.readColumns = readColumns; + this.outputSchema = outputSchema; + this.readColumnIndexes = indexes(readColumns); + this.outputIndexes = outputIndexes(outputColumns, readColumnIndexes); + RowRange range = request.rowRange().orElse(null); + this.rangeStart = range == null ? 0 : range.startInclusive(); + this.rangeEnd = range == null ? Long.MAX_VALUE : range.endExclusive(); + this.limit = request.limit().isPresent() ? request.limit().getAsLong() : Long.MAX_VALUE; + this.rowGroups = rowGroups(fileReader.getRowGroups()); + } + + @Override + public boolean next() throws IOException { + if (closed || exhausted) { + current = null; + return false; + } + if (emitted == limit) { Review Comment: Two issues: (1) for limit > 0 this branch is unreachable — reaching the limit inside the batch loop already sets exhausted = true, so the `closed || exhausted` check above returns false first; the only reachable case is limit == 0. (2) When the limit is reached mid-scan, next() sets exhausted without closing the ParquetFileReader or any open pages (unlike natural exhaustion, which goes through finish()), so after next() returns false the cursor is in different release states depending on the exit path; a caller that stops at the first false return without calling close() leaks the file handle. Suggest closing the reader when the limit is reached while keeping `current` retrievable so batch() stays valid after the final true return. ########## maven-projects/io-parquet/pom.xml: ########## @@ -0,0 +1,119 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + + 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. + +--> + +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.graphar</groupId> + <artifactId>graphar-root</artifactId> + <version>${graphar.version}</version> + <relativePath>../pom.xml</relativePath> + </parent> + + <artifactId>graphar-io-parquet</artifactId> + <packaging>jar</packaging> + <version>${graphar.version}</version> + + <name>graphar-io-parquet</name> + + <properties> + <maven.compiler.source>11</maven.compiler.source> + <maven.compiler.target>11</maven.compiler.target> + <hadoop.version>3.3.0</hadoop.version> Review Comment: The latest is 3.5.0 (https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-common/3.5.0), what was the reason of pinning 3.3.0? ########## maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetOutputFile.java: ########## @@ -0,0 +1,93 @@ +/* + * 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.graphar.io.parquet; + +import java.io.IOException; +import java.util.Objects; +import org.apache.graphar.storage.OutputFile; +import org.apache.graphar.storage.PositionOutput; +import org.apache.parquet.io.PositionOutputStream; + +/** Adapts a GraphAr storage output file to Parquet's output-file interface. */ +final class ParquetOutputFile implements org.apache.parquet.io.OutputFile { + private final OutputFile outputFile; + + ParquetOutputFile(OutputFile outputFile) { + this.outputFile = Objects.requireNonNull(outputFile, "outputFile"); + } + + @Override + public PositionOutputStream create(long blockSizeHint) throws IOException { + return new ParquetPositionOutputStream(outputFile.create()); + } + + @Override + public PositionOutputStream createOrOverwrite(long blockSizeHint) throws IOException { + return new ParquetPositionOutputStream(outputFile.createOrOverwrite()); + } + + @Override + public boolean supportsBlockSize() { + return false; + } + + @Override + public long defaultBlockSize() { + return 0; + } + + @Override + public String getPath() { + return outputFile.uri().toString(); + } + + private static final class ParquetPositionOutputStream extends PositionOutputStream { + private final PositionOutput output; + + private ParquetPositionOutputStream(PositionOutput output) { + this.output = output; + } + + @Override + public long getPos() throws IOException { + return output.position(); + } + + @Override + public void write(int value) throws IOException { + output.write(new byte[] {(byte) value}); Review Comment: This allocates a fresh one-byte array on every single-byte write. Parquet serializes many small values (page headers, variant lengths, footer metadata) through write(int), producing avoidable garbage per file. A reusable one-byte buffer is safe here because the writer stream is used by a single thread. ########## maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetBatchCursor.java: ########## @@ -0,0 +1,350 @@ +/* + * 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.graphar.io.parquet; + +import java.io.IOException; +import java.time.Instant; +import java.time.LocalDate; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.graphar.io.BatchCursor; +import org.apache.graphar.io.ReadRequest; +import org.apache.graphar.io.RecordBatch; +import org.apache.graphar.io.RowRange; +import org.apache.graphar.io.Schema; +import org.apache.graphar.io.ValueVector; +import org.apache.graphar.io.VectorRecordBatch; +import org.apache.parquet.column.page.PageReadStore; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.convert.GroupRecordConverter; +import org.apache.parquet.filter2.columnindex.RowRanges; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.internal.filter2.columnindex.ColumnIndexStore.MissingOffsetIndexException; +import org.apache.parquet.io.ColumnIOFactory; +import org.apache.parquet.io.MessageColumnIO; +import org.apache.parquet.io.RecordReader; +import org.apache.parquet.schema.MessageType; + +/** Streams materialized Parquet row groups as neutral record batches. */ +final class ParquetBatchCursor implements BatchCursor { + private static final int BATCH_ROWS = 1_024; + private final ParquetFileReader fileReader; + private final MessageType fileSchema; + private final MessageType readSchema; + private final List<ParquetColumn> readColumns; + private final Schema outputSchema; + private final Map<String, Integer> readColumnIndexes; + private final int[] outputIndexes; + private final long rangeStart; + private final long rangeEnd; + private final long limit; + private final List<BlockRange> rowGroups; + private int nextRowGroup; + private long emitted; + private long rowsRemainingInGroup; + private PageReadStore pages; + private RecordReader<Group> rows; + private boolean exhausted; + private boolean closed; + private RecordBatch current; + + ParquetBatchCursor( + ParquetFileReader fileReader, + MessageType fileSchema, + MessageType readSchema, + List<ParquetColumn> readColumns, + List<ParquetColumn> outputColumns, + Schema outputSchema, + ReadRequest request) { + this.fileReader = fileReader; + this.fileSchema = fileSchema; + this.readSchema = readSchema; + this.readColumns = readColumns; + this.outputSchema = outputSchema; + this.readColumnIndexes = indexes(readColumns); + this.outputIndexes = outputIndexes(outputColumns, readColumnIndexes); + RowRange range = request.rowRange().orElse(null); + this.rangeStart = range == null ? 0 : range.startInclusive(); + this.rangeEnd = range == null ? Long.MAX_VALUE : range.endExclusive(); + this.limit = request.limit().isPresent() ? request.limit().getAsLong() : Long.MAX_VALUE; + this.rowGroups = rowGroups(fileReader.getRowGroups()); + } + + @Override + public boolean next() throws IOException { + if (closed || exhausted) { + current = null; + return false; + } + if (emitted == limit) { + finish(); + return false; + } + try { + while (true) { + if (rows == null && !openNextRowGroup()) { + finish(); + return false; + } + int batchSize = + (int) Math.min(Math.min(rowsRemainingInGroup, BATCH_ROWS), limit - emitted); + Object[][] columns = new Object[outputIndexes.length][batchSize]; + for (int index = 0; index < batchSize; index++) { + Group group = rows.read(); + project(values(group), columns, index); + emitted++; + rowsRemainingInGroup--; + } + if (rowsRemainingInGroup == 0) { + closePages(); + } + current = batch(columns, batchSize); + if (emitted == limit) exhausted = true; + return true; + } + } catch (MissingOffsetIndexException exception) { + try { + closeReader(); + } catch (IOException closeException) { + exception.addSuppressed(closeException); + } + throw new UnsupportedOperationException( + "Physical Parquet row ranges require an Offset Index; refusing JVM fallback.", + exception); + } catch (IOException | RuntimeException exception) { + try { + closeReader(); + } catch (IOException closeException) { + exception.addSuppressed(closeException); + } + throw exception; + } + } + + @Override + public RecordBatch batch() { + if (current == null) { + throw new IllegalStateException("No current batch. Call next() before batch()."); + } + return current; + } + + @Override + public void close() throws IOException { + current = null; + exhausted = true; + closeReader(); + } + + private Object[] values(Group group) { + Object[] values = new Object[readColumns.size()]; + for (int index = 0; index < readColumns.size(); index++) { + if (group.getFieldRepetitionCount(index) != 0) { + values[index] = value(group, index, readColumns.get(index)); + } + } + return values; + } + + private static Object value(Group group, int index, ParquetColumn column) { + if (column.field().type().kind() == org.apache.graphar.io.ColumnType.Kind.LIST) { + return listValue(group, index, column); + } + return scalarValue(group, index, column.field().type().kind()); + } + + private static List<Object> listValue(Group group, int index, ParquetColumn column) { + Group list = group.getGroup(index, 0); + if (list.getType().getFieldCount() != 1) { + throw new IllegalArgumentException( + "Unsupported Parquet LIST field: " + column.field().name()); + } + int count = list.getFieldRepetitionCount(0); + if (count == 0) { + return List.of(); + } + List<Object> values = new ArrayList<>(count); + org.apache.graphar.io.ColumnType element = + column.field().type().elementType().orElseThrow(); + for (int elementIndex = 0; elementIndex < count; elementIndex++) { + Group elementGroup = list.getGroup(0, elementIndex); Review Comment: The variable names are inverted relative to the Parquet spec terminology, which makes the element-extraction logic hard to verify: `list` holds the LIST group instance (the outer group), while `elementGroup` actually holds one instance of the middle repeated "list" wrapper group, not the element itself. The logic reads correctly (each wrapper instance contains one primitive element), but the naming suggests the primitive element is being treated as a group. ########## maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetBatchCursor.java: ########## @@ -0,0 +1,350 @@ +/* + * 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.graphar.io.parquet; + +import java.io.IOException; +import java.time.Instant; +import java.time.LocalDate; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.graphar.io.BatchCursor; +import org.apache.graphar.io.ReadRequest; +import org.apache.graphar.io.RecordBatch; +import org.apache.graphar.io.RowRange; +import org.apache.graphar.io.Schema; +import org.apache.graphar.io.ValueVector; +import org.apache.graphar.io.VectorRecordBatch; +import org.apache.parquet.column.page.PageReadStore; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.convert.GroupRecordConverter; +import org.apache.parquet.filter2.columnindex.RowRanges; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.internal.filter2.columnindex.ColumnIndexStore.MissingOffsetIndexException; +import org.apache.parquet.io.ColumnIOFactory; +import org.apache.parquet.io.MessageColumnIO; +import org.apache.parquet.io.RecordReader; +import org.apache.parquet.schema.MessageType; + +/** Streams materialized Parquet row groups as neutral record batches. */ +final class ParquetBatchCursor implements BatchCursor { + private static final int BATCH_ROWS = 1_024; + private final ParquetFileReader fileReader; + private final MessageType fileSchema; + private final MessageType readSchema; + private final List<ParquetColumn> readColumns; + private final Schema outputSchema; + private final Map<String, Integer> readColumnIndexes; + private final int[] outputIndexes; + private final long rangeStart; + private final long rangeEnd; + private final long limit; + private final List<BlockRange> rowGroups; + private int nextRowGroup; + private long emitted; + private long rowsRemainingInGroup; + private PageReadStore pages; + private RecordReader<Group> rows; + private boolean exhausted; + private boolean closed; + private RecordBatch current; + + ParquetBatchCursor( + ParquetFileReader fileReader, + MessageType fileSchema, + MessageType readSchema, + List<ParquetColumn> readColumns, + List<ParquetColumn> outputColumns, + Schema outputSchema, + ReadRequest request) { + this.fileReader = fileReader; + this.fileSchema = fileSchema; + this.readSchema = readSchema; + this.readColumns = readColumns; + this.outputSchema = outputSchema; + this.readColumnIndexes = indexes(readColumns); + this.outputIndexes = outputIndexes(outputColumns, readColumnIndexes); + RowRange range = request.rowRange().orElse(null); + this.rangeStart = range == null ? 0 : range.startInclusive(); + this.rangeEnd = range == null ? Long.MAX_VALUE : range.endExclusive(); + this.limit = request.limit().isPresent() ? request.limit().getAsLong() : Long.MAX_VALUE; + this.rowGroups = rowGroups(fileReader.getRowGroups()); + } + + @Override + public boolean next() throws IOException { + if (closed || exhausted) { + current = null; + return false; + } + if (emitted == limit) { + finish(); + return false; + } + try { + while (true) { + if (rows == null && !openNextRowGroup()) { + finish(); + return false; + } + int batchSize = + (int) Math.min(Math.min(rowsRemainingInGroup, BATCH_ROWS), limit - emitted); + Object[][] columns = new Object[outputIndexes.length][batchSize]; + for (int index = 0; index < batchSize; index++) { + Group group = rows.read(); + project(values(group), columns, index); + emitted++; + rowsRemainingInGroup--; + } + if (rowsRemainingInGroup == 0) { + closePages(); + } + current = batch(columns, batchSize); + if (emitted == limit) exhausted = true; + return true; + } + } catch (MissingOffsetIndexException exception) { + try { + closeReader(); + } catch (IOException closeException) { + exception.addSuppressed(closeException); + } + throw new UnsupportedOperationException( + "Physical Parquet row ranges require an Offset Index; refusing JVM fallback.", + exception); + } catch (IOException | RuntimeException exception) { + try { + closeReader(); + } catch (IOException closeException) { + exception.addSuppressed(closeException); + } + throw exception; + } + } + + @Override + public RecordBatch batch() { + if (current == null) { + throw new IllegalStateException("No current batch. Call next() before batch()."); + } + return current; + } + + @Override + public void close() throws IOException { + current = null; + exhausted = true; + closeReader(); + } + + private Object[] values(Group group) { + Object[] values = new Object[readColumns.size()]; + for (int index = 0; index < readColumns.size(); index++) { + if (group.getFieldRepetitionCount(index) != 0) { + values[index] = value(group, index, readColumns.get(index)); + } + } + return values; + } + + private static Object value(Group group, int index, ParquetColumn column) { + if (column.field().type().kind() == org.apache.graphar.io.ColumnType.Kind.LIST) { + return listValue(group, index, column); + } + return scalarValue(group, index, column.field().type().kind()); + } + + private static List<Object> listValue(Group group, int index, ParquetColumn column) { + Group list = group.getGroup(index, 0); + if (list.getType().getFieldCount() != 1) { + throw new IllegalArgumentException( + "Unsupported Parquet LIST field: " + column.field().name()); + } + int count = list.getFieldRepetitionCount(0); + if (count == 0) { + return List.of(); + } + List<Object> values = new ArrayList<>(count); + org.apache.graphar.io.ColumnType element = + column.field().type().elementType().orElseThrow(); + for (int elementIndex = 0; elementIndex < count; elementIndex++) { + Group elementGroup = list.getGroup(0, elementIndex); + if (elementGroup.getType().getFieldCount() != 1 + || elementGroup.getFieldRepetitionCount(0) != 1) { + throw new IllegalArgumentException( + "Unsupported Parquet LIST element: " + column.field().name()); + } + values.add(scalarValue(elementGroup, 0, element, 0)); + } + return Collections.unmodifiableList(values); + } + + private static Object scalarValue( + Group group, int index, org.apache.graphar.io.ColumnType.Kind kind) { + return scalarValue(group, index, org.apache.graphar.io.ColumnType.of(kind), 0); + } + + private static Object scalarValue( + Group group, int index, org.apache.graphar.io.ColumnType type, int repetitionIndex) { + switch (type.kind()) { + case BOOLEAN: + return group.getBoolean(index, repetitionIndex); + case INT8: + return (byte) group.getInteger(index, repetitionIndex); + case INT16: + return (short) group.getInteger(index, repetitionIndex); + case INT32: + return group.getInteger(index, repetitionIndex); + case INT64: + return group.getLong(index, repetitionIndex); + case FLOAT32: + return group.getFloat(index, repetitionIndex); + case FLOAT64: + return group.getDouble(index, repetitionIndex); + case STRING: + return group.getBinary(index, repetitionIndex).toStringUsingUTF8(); + case BINARY: + return group.getBinary(index, repetitionIndex).getBytes(); + case DATE: + return LocalDate.ofEpochDay(group.getInteger(index, repetitionIndex)); + case TIMESTAMP_MILLIS: + return Instant.ofEpochMilli(group.getLong(index, repetitionIndex)); + default: + throw new IllegalArgumentException( + "Unsupported Parquet column type: " + type.kind()); + } + } + + private void project(Object[] values, Object[][] columns, int row) { + for (int index = 0; index < outputIndexes.length; index++) { + Object value = values[outputIndexes[index]]; + columns[index][row] = value instanceof byte[] ? ((byte[]) value).clone() : value; + } + } + + private RecordBatch batch(Object[][] columns, int rowCount) { + List<ValueVector> vectors = new ArrayList<>(columns.length); + for (int index = 0; index < columns.length; index++) { + vectors.add(new ParquetValueVector(outputSchema.fields().get(index), columns[index])); + } + return new VectorRecordBatch(outputSchema, vectors, rowCount); + } + + private void finish() throws IOException { + exhausted = true; + current = null; + closeReader(); + } + + private void closeReader() throws IOException { + if (!closed) { + closed = true; + try { + closePages(); + } finally { + fileReader.close(); + } + } + } + + private boolean openNextRowGroup() throws IOException { + BlockRange rowGroup = nextRange(); + if (rowGroup == null) return false; + pages = Review Comment: Row-group reads always go through `readFilteredRowGroup` with a RowRanges object, which requires offset indexes in the file. When no row range was requested (rangeStart=0, rangeEnd=MAX_VALUE) or the range covers the whole group, this still forces the offset-index code path, so a plain full scan of a Parquet file without offset indexes (older or foreign writers) throws `MissingOffsetIndexException` and surfaces as `UnsupportedOperationException` — even though no filtering is needed. The catch message "Physical Parquet row ranges require an Offset Index" is misleading for the same reason. There is no non-filtered fallback in this module; suggest calling `fileReader.readRowGroup(rowGroup.index)` when the selection covers the entire row group (a flag computed in nextRange(), where the absolute bounds are known). ########## maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetPhysicalReader.java: ########## @@ -0,0 +1,426 @@ +/* + * 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.graphar.io.parquet; + +import java.io.IOException; +import java.net.URI; +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.graphar.io.ColumnRef; +import org.apache.graphar.io.ColumnType; +import org.apache.graphar.io.Field; +import org.apache.graphar.io.PhysicalReader; +import org.apache.graphar.io.ReadCapability; +import org.apache.graphar.io.ReadReport; +import org.apache.graphar.io.ReadRequest; +import org.apache.graphar.io.ReadResult; +import org.apache.graphar.io.Schema; +import org.apache.graphar.storage.InputFile; +import org.apache.graphar.storage.Storage; +import org.apache.parquet.ParquetReadOptions; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.io.SeekableInputStream; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Type; + +/** A storage-backed reader for GraphAr primitive and LIST Parquet fields. */ +public final class ParquetPhysicalReader implements PhysicalReader { + private static final Set<ReadCapability> CAPABILITIES = + Collections.unmodifiableSet( + EnumSet.of( + ReadCapability.PROJECTION, + ReadCapability.ROW_RANGE, + ReadCapability.LIMIT)); + + private static final int DEFAULT_FOOTER_CACHE_CAPACITY = 256; + + private final Storage storage; + private final FooterCache footers; + + /** + * Creates a reader that resolves each request URI through {@code storage} and remembers the + * footers of recently read files. + */ + public ParquetPhysicalReader(Storage storage) { + this(storage, DEFAULT_FOOTER_CACHE_CAPACITY); + } + + /** + * Creates a reader that keeps at most {@code footerCacheCapacity} Parquet footers in memory. + * GraphAr chunk files are immutable once published, so repeated range reads of one chunk parse + * its footer once. A capacity of zero parses the footer on every request. + * + * @param storage resolves each request URI to a readable file + * @param footerCacheCapacity maximum number of remembered footers + */ + public ParquetPhysicalReader(Storage storage, int footerCacheCapacity) { + this.storage = Objects.requireNonNull(storage, "storage"); + this.footers = new FooterCache(footerCacheCapacity); + } + + @Override + public Set<ReadCapability> capabilities() { + return CAPABILITIES; + } + + @Override + public ReadResult read(ReadRequest request) throws IOException { + Objects.requireNonNull(request, "request"); + if (!request.filters().isEmpty()) { + throw new UnsupportedOperationException( + "Parquet filter pushdown is not implemented; refusing semantic fallback."); + } + InputFile inputFile = + Objects.requireNonNull(storage.inputFile(request.uri()), "storage inputFile"); + ParquetFileReader fileReader = null; + try { + fileReader = open(request.uri(), new ParquetInputFile(inputFile)); + MessageType fileSchema = fileReader.getFooter().getFileMetaData().getSchema(); + List<ParquetColumn> fileColumns = columns(fileSchema); + Map<String, ParquetColumn> columnsByName = byName(fileColumns); + List<ParquetColumn> outputColumns = outputColumns(request, fileColumns, columnsByName); + List<ParquetColumn> readColumns = readColumns(fileColumns, outputColumns); + MessageType readSchema = + new MessageType(fileSchema.getName(), parquetTypes(readColumns)); + fileReader.setRequestedSchema(readSchema); + + Schema outputSchema = new Schema(fields(outputColumns)); + ReadReport report = new ReadReport(applied(request), declined(request)); + ParquetBatchCursor cursor = + new ParquetBatchCursor( + fileReader, + fileSchema, + readSchema, + readColumns, + outputColumns, + outputSchema, + request); + fileReader = null; + return new ReadResult(request, cursor, report); + } finally { + if (fileReader != null) { + fileReader.close(); + } + } + } + + /** + * Opens a Parquet reader, reusing a remembered footer when the file is unchanged in size. The + * returned reader owns the stream opened here and closes it. + */ + private ParquetFileReader open(URI uri, ParquetInputFile file) throws IOException { + ParquetReadOptions options = ParquetReadOptions.builder().build(); + long size = file.getLength(); + ParquetMetadata remembered = footers.get(uri, size); Review Comment: The footer cache keys entries only by (uri, file size), so a file overwritten in place with different content of the same length is parsed with a stale footer — wrong schema, wrong row-group offsets, corrupt reads. The "chunk files are immutable" javadoc assumption is not enforced: the storage API exposes `OutputFile.createOrOverwrite()` (storage-api, storage-local, storage-s3) and this module's own writer accepts `WriteMode.OVERWRITE`. Consider invalidating cache entries on write paths, keying on a stronger signal, or at minimum documenting the immutability contract where the cache is defined. -- 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]
