keksmd commented on code in PR #975: URL: https://github.com/apache/incubator-graphar/pull/975#discussion_r4094042800
########## 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: Fixed in eb0ad12. List values are copied with a null-tolerant unmodifiable copy; a null element is accepted when the element field is nullable and rejected with a message naming the column and index otherwise. ########## 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: Fixed in eb0ad12. List values are snapshotted recursively, so nested lists are unmodifiable copies at every depth (`snapshotsNestedListsAtEveryDepth`). ########## 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: Fixed in eb0ad12. `ObjectValueVector` now checks every value at construction: null in a non-nullable field, Java type per kind (documented in the class javadoc), integer range for INT8/16/32, and list elements recursively against the element field. Both `RecordBatches` factories go through it and document the `IllegalArgumentException`. ########## 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: Fixed in eb0ad12, reader and writer together. The writer emits the 3-level layout with a REQUIRED or OPTIONAL `element` following the element field's nullability; the reader accepts REQUIRED/OPTIONAL elements and maps an absent element to null. Tests: a file written directly with a Spark-style schema (`optional int64 element`, with a null entry) reads back, and our own files are asserted to carry the spec layout. -- 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]
