This is an automated email from the ASF dual-hosted git repository.
SemyonSinchenko pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-graphar.git
The following commit(s) were added to refs/heads/main by this push:
new d7fe5589 feat(java): read and write Parquet through column vectors
(#975)
d7fe5589 is described below
commit d7fe5589694d23e2d30f22b9853c3fca94666131
Author: alex <[email protected]>
AuthorDate: Fri Oct 9 18:06:28 2026 +0300
feat(java): read and write Parquet through column vectors (#975)
* feat(java): read and write Parquet through column vectors
Adds the graphar-io-parquet module: a physical reader that materializes each
Parquet row group as ValueVector columns inside a VectorRecordBatch,
applying
projection and half-open row ranges and reporting what it pushed down, and a
physical writer that emits Offset Indexes with bounded pages.
Binary values are exposed as read-only ByteBuffers over a private copy and
list
values as immutable lists, so a consumer cannot alias the decoder's buffers.
Parquet cannot append to a closed file, so WriteMode.APPEND is rejected with
UnsupportedOperationException instead of being downgraded to an overwrite.
io-api gains ObjectValueVector and RecordBatches, an in-memory vector and a
row-to-column transposer so producers can build batches without Arrow.
Closes #964
* fix(java): address review of the Parquet column-vector backend
Follow the Parquet LIST spec: write and accept 3-level lists whose element
is REQUIRED or OPTIONAL, carrying null elements, so files from Spark, Arrow
and other writers read and our files are standard.
Validate producer values in ObjectValueVector against the field type and
nullability, naming column and index, and snapshot list values recursively
without rejecting null elements.
Read whole row groups without the offset-index path, release the file as
soon as the limit is reached, make the footer cache opt-in with an explicit
immutability contract, reuse the single-byte write buffer, and replace
hadoop-common and hadoop-mapreduce-client-core with the shaded Hadoop
client 3.4.2 (runtime jars 135 -> 19; 3.5.x targets Java 17).
---
.github/workflows/java-sdk.yml | 2 +-
.../org/apache/graphar/io/ObjectValueVector.java | 207 +++++++++
.../java/org/apache/graphar/io/RecordBatches.java | 85 ++++
.../org/apache/graphar/io/RecordBatchesTest.java | 165 +++++++
maven-projects/io-parquet/pom.xml | 120 +++++
.../graphar/io/parquet/ParquetBatchCursor.java | 363 ++++++++++++++++
.../apache/graphar/io/parquet/ParquetColumn.java | 42 ++
.../graphar/io/parquet/ParquetInputFile.java | 44 ++
.../graphar/io/parquet/ParquetOutputFile.java | 95 ++++
.../graphar/io/parquet/ParquetPhysicalReader.java | 433 ++++++++++++++++++
.../graphar/io/parquet/ParquetPhysicalWriter.java | 291 +++++++++++++
.../io/parquet/ParquetSeekableInputStream.java | 88 ++++
.../graphar/io/parquet/ParquetValueVector.java | 58 +++
.../io/parquet/ParquetListRoundTripTest.java | 116 +++++
.../parquet/ParquetPhysicalIoEfficiencyTest.java | 483 +++++++++++++++++++++
.../parquet/ParquetPhysicalReaderFixtureTest.java | 121 ++++++
.../io/parquet/ParquetPhysicalWriterModeTest.java | 122 ++++++
.../ParquetPhysicalWriterOffsetIndexTest.java | 166 +++++++
.../io/parquet/ParquetSpecConformanceTest.java | 292 +++++++++++++
maven-projects/pom.xml | 1 +
20 files changed, 3293 insertions(+), 1 deletion(-)
diff --git a/.github/workflows/java-sdk.yml b/.github/workflows/java-sdk.yml
index 5a45ac49..f2ff2feb 100644
--- a/.github/workflows/java-sdk.yml
+++ b/.github/workflows/java-sdk.yml
@@ -55,7 +55,7 @@ jobs:
test:
runs-on: ubuntu-latest
env:
- SDK_MODULES: storage-api,storage-local,io-api,storage-s3,core
+ SDK_MODULES: storage-api,storage-local,io-api,io-parquet,storage-s3,core
steps:
- uses: actions/checkout@v7
with:
diff --git
a/maven-projects/io-api/src/main/java/org/apache/graphar/io/ObjectValueVector.java
b/maven-projects/io-api/src/main/java/org/apache/graphar/io/ObjectValueVector.java
new file mode 100644
index 00000000..ec2d7916
--- /dev/null
+++
b/maven-projects/io-api/src/main/java/org/apache/graphar/io/ObjectValueVector.java
@@ -0,0 +1,207 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.LocalDate;
+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. Every value is checked against the vector {@link
Field} at
+ * construction, so a malformed value fails here, naming the column and index,
instead of later
+ * inside a format encoder. The accepted Java representations are:
+ *
+ * <ul>
+ * <li>{@code BOOLEAN}: {@link Boolean}
+ * <li>{@code INT8}, {@code INT16}, {@code INT32}, {@code INT64}: {@link
Byte}, {@link Short},
+ * {@link Integer} or {@link Long} whose value fits the kind
+ * <li>{@code FLOAT32}, {@code FLOAT64}: {@link Float} or {@link Double}
+ * <li>{@code STRING}: {@link String}
+ * <li>{@code BINARY}: {@link ByteBuffer}
+ * <li>{@code DATE}: {@link LocalDate}
+ * <li>{@code TIMESTAMP_MILLIS}: {@link Instant}
+ * <li>{@code LIST}: {@link List} whose elements follow the element field,
including its
+ * nullability
+ * </ul>
+ *
+ * <p>Other kinds are checked for nullability only. List values are
snapshotted recursively into
+ * unmodifiable copies that may hold null elements, so neither the producer
nor a consumer can
+ * mutate the vector afterwards at any nesting depth. Binary values are held
as given.
+ */
+public final class ObjectValueVector implements ValueVector {
+ private final Field field;
+ private final List<Object> values;
+
+ /**
+ * Creates a vector over {@code values}.
+ *
+ * @throws IllegalArgumentException if a value is null in a non-nullable
field or does not have
+ * a Java type accepted for the field type
+ */
+ 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 (int index = 0; index < values.size(); index++) {
+ copy.add(checked(field, values.get(index), field.name(), index));
+ }
+ this.values = Collections.unmodifiableList(copy);
+ }
+
+ @Override
+ public Field field() {
+ return field;
+ }
+
+ @Override
+ public int valueCount() {
+ return values.size();
+ }
+
+ @Override
+ public boolean isNull(int index) {
+ return values.get(index) == null;
+ }
+
+ @Override
+ public Object getObject(int index) {
+ return values.get(index);
+ }
+
+ private static Object checked(Field field, Object value, String column,
int index) {
+ if (value == null) {
+ if (!field.nullable()) {
+ throw new IllegalArgumentException(
+ "Column '"
+ + column
+ + "' index "
+ + index
+ + ": null in a non-nullable field");
+ }
+ return null;
+ }
+ ColumnType type = field.type();
+ switch (type.kind()) {
+ case BOOLEAN:
+ return require(value, column, index, type, Boolean.class);
+ case INT8:
+ case INT16:
+ case INT32:
+ case INT64:
+ return integer(
+ type,
+ require(
+ value,
+ column,
+ index,
+ type,
+ Byte.class,
+ Short.class,
+ Integer.class,
+ Long.class),
+ column,
+ index);
+ case FLOAT32:
+ case FLOAT64:
+ return require(value, column, index, type, Float.class,
Double.class);
+ case STRING:
+ return require(value, column, index, type, String.class);
+ case BINARY:
+ return require(value, column, index, type, ByteBuffer.class);
+ case DATE:
+ return require(value, column, index, type, LocalDate.class);
+ case TIMESTAMP_MILLIS:
+ return require(value, column, index, type, Instant.class);
+ case LIST:
+ return list(type, require(value, column, index, type,
List.class), column, index);
+ default:
+ return value;
+ }
+ }
+
+ private static Number integer(ColumnType type, Number value, String
column, int index) {
+ long bound;
+ switch (type.kind()) {
+ case INT8:
+ bound = Byte.MAX_VALUE;
+ break;
+ case INT16:
+ bound = Short.MAX_VALUE;
+ break;
+ case INT32:
+ bound = Integer.MAX_VALUE;
+ break;
+ default:
+ return value;
+ }
+ long actual = value.longValue();
+ if (actual > bound || actual < -bound - 1) {
+ throw new IllegalArgumentException(
+ "Column '"
+ + column
+ + "' index "
+ + index
+ + ": "
+ + actual
+ + " is out of range for "
+ + type.kind());
+ }
+ return value;
+ }
+
+ private static List<Object> list(ColumnType type, List<?> value, String
column, int index) {
+ Field element = type.children().get(0);
+ List<Object> copy = new ArrayList<>(value.size());
+ for (Object elementValue : value) {
+ copy.add(checked(element, elementValue, column, index));
+ }
+ return Collections.unmodifiableList(copy);
+ }
+
+ @SafeVarargs
+ private static <T> T require(
+ Object value,
+ String column,
+ int index,
+ ColumnType type,
+ Class<? extends T>... accepted) {
+ for (Class<? extends T> candidate : accepted) {
+ if (candidate.isInstance(value)) {
+ return candidate.cast(value);
+ }
+ }
+ throw new IllegalArgumentException(
+ "Column '"
+ + column
+ + "' index "
+ + index
+ + ": "
+ + value.getClass().getSimpleName()
+ + " is not a "
+ + type.kind()
+ + " value");
+ }
+}
diff --git
a/maven-projects/io-api/src/main/java/org/apache/graphar/io/RecordBatches.java
b/maven-projects/io-api/src/main/java/org/apache/graphar/io/RecordBatches.java
new file mode 100644
index 00000000..25102cd8
--- /dev/null
+++
b/maven-projects/io-api/src/main/java/org/apache/graphar/io/RecordBatches.java
@@ -0,0 +1,85 @@
+/*
+ * 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
+ * @throws IllegalArgumentException if a row has the wrong width, or a
value is null in a
+ * non-nullable field or has a Java type the field does not accept;
see {@link
+ * ObjectValueVector}
+ */
+ public static RecordBatch ofRows(Schema schema, List<? extends List<?>>
rows) {
+ Objects.requireNonNull(schema, "A batch schema cannot be null.");
+ Objects.requireNonNull(rows, "Batch rows cannot be null.");
+ int width = schema.fields().size();
+ List<List<Object>> valuesByColumn = new ArrayList<>(width);
+ for (int column = 0; column < width; column++) {
+ valuesByColumn.add(new ArrayList<>(rows.size()));
+ }
+ for (List<?> row : rows) {
+ Objects.requireNonNull(row, "A batch row cannot be null.");
+ if (row.size() != width) {
+ throw new IllegalArgumentException("A batch row does not match
the schema width.");
+ }
+ for (int column = 0; column < width; column++) {
+ valuesByColumn.get(column).add(row.get(column));
+ }
+ }
+ List<ValueVector> columns = new ArrayList<>(width);
+ for (int column = 0; column < width; column++) {
+ columns.add(
+ new ObjectValueVector(schema.fields().get(column),
valuesByColumn.get(column)));
+ }
+ return new VectorRecordBatch(schema, columns, rows.size());
+ }
+
+ /**
+ * Transposes row-major arrays into a columnar batch.
+ *
+ * @param schema the batch schema; every row must hold one value per field
+ * @param rows the rows, each an array of boxed values in schema order
+ * @return a batch backed by {@link ObjectValueVector} columns
+ * @throws IllegalArgumentException under the same conditions as {@link
#ofRows(Schema, List)}
+ */
+ public static RecordBatch ofArrays(Schema schema, List<Object[]> rows) {
+ Objects.requireNonNull(rows, "Batch rows cannot be null.");
+ List<List<Object>> lists = new ArrayList<>(rows.size());
+ for (Object[] row : rows) {
+ Objects.requireNonNull(row, "A batch row cannot be null.");
+ List<Object> list = new ArrayList<>(row.length);
+ for (Object value : row) {
+ list.add(value);
+ }
+ lists.add(list);
+ }
+ return ofRows(schema, lists);
+ }
+}
diff --git
a/maven-projects/io-api/src/test/java/org/apache/graphar/io/RecordBatchesTest.java
b/maven-projects/io-api/src/test/java/org/apache/graphar/io/RecordBatchesTest.java
new file mode 100644
index 00000000..5ef8b8a4
--- /dev/null
+++
b/maven-projects/io-api/src/test/java/org/apache/graphar/io/RecordBatchesTest.java
@@ -0,0 +1,165 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.junit.Test;
+
+public class RecordBatchesTest {
+ private static final Schema SCHEMA =
+ new Schema(
+ List.of(
+ new Field("id",
ColumnType.of(ColumnType.Kind.INT64), false),
+ new Field("name",
ColumnType.of(ColumnType.Kind.STRING), true)));
+
+ @Test
+ public void transposesRowsIntoColumns() {
+ RecordBatch batch =
+ RecordBatches.ofRows(
+ SCHEMA, List.of(Arrays.asList(1L, "a"),
Arrays.asList(2L, null)));
+
+ assertEquals(2, batch.rowCount());
+ assertEquals(2, batch.columnCount());
+ assertEquals(1L, batch.column(0).getObject(0));
+ assertEquals(2L, batch.column(0).getObject(1));
+ assertEquals("a", batch.column(1).getObject(0));
+ assertTrue(batch.column(1).isNull(1));
+ assertFalse(batch.column(0).isNull(1));
+ assertEquals(SCHEMA.fields().get(1), batch.column(1).field());
+ }
+
+ @Test
+ public void acceptsArrays() {
+ RecordBatch batch =
+ RecordBatches.ofArrays(SCHEMA, List.<Object[]>of(new Object[]
{1L, "a"}));
+
+ assertEquals(1, batch.rowCount());
+ assertEquals("a", batch.column(1).getObject(0));
+ }
+
+ @Test
+ public void buildsAnEmptyBatch() {
+ RecordBatch batch = RecordBatches.ofRows(SCHEMA, List.of());
+
+ assertEquals(0, batch.rowCount());
+ assertEquals(0, batch.column(0).valueCount());
+ }
+
+ @Test
+ public void rejectsARowOfTheWrongWidth() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> RecordBatches.ofRows(SCHEMA, List.of(List.of(1L))));
+ }
+
+ @Test
+ public void copiesListValuesOnRead() {
+ Field field =
+ new Field("tags",
ColumnType.listOf(ColumnType.of(ColumnType.Kind.STRING)), true);
+ List<Object> tags = new ArrayList<>(List.of("x"));
+ ObjectValueVector vector = new ObjectValueVector(field, List.of(tags));
+ tags.add("y");
+
+ assertEquals(List.of("x"), vector.getObject(0));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> ((List<Object>) vector.getObject(0)).add("z"));
+ }
+
+ @Test
+ public void keepsNullElementsOfANullableElementList() {
+ Field field =
+ new Field("tags",
ColumnType.listOf(ColumnType.of(ColumnType.Kind.STRING)), true);
+
+ ObjectValueVector vector = new ObjectValueVector(field,
List.of(Arrays.asList("x", null)));
+
+ assertEquals(Arrays.asList("x", null), vector.getObject(0));
+ }
+
+ @Test
+ public void rejectsANullElementOfARequiredElementList() {
+ Field field =
+ new Field(
+ "tags",
+ ColumnType.listOfElement(
+ new Field("element",
ColumnType.of(ColumnType.Kind.STRING), false)),
+ true);
+
+ IllegalArgumentException error =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> new ObjectValueVector(field,
List.of(Arrays.asList("x", null))));
+ assertTrue(error.getMessage(), error.getMessage().contains("'tags'
index 0"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void snapshotsNestedListsAtEveryDepth() {
+ Field field =
+ new Field(
+ "matrix",
+
ColumnType.listOf(ColumnType.listOf(ColumnType.of(ColumnType.Kind.INT64))),
+ true);
+ List<Object> inner = new ArrayList<>(List.of(1L));
+ ObjectValueVector vector = new ObjectValueVector(field,
List.of(List.of(inner)));
+ inner.add(2L);
+
+ List<Object> outer = (List<Object>) vector.getObject(0);
+ assertEquals(List.of(List.of(1L)), outer);
+ assertThrows(
+ UnsupportedOperationException.class, () -> ((List<Object>)
outer.get(0)).add(3L));
+ }
+
+ @Test
+ public void rejectsANullInANonNullableColumn() {
+ IllegalArgumentException error =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ RecordBatches.ofRows(
+ SCHEMA,
+ List.of(Arrays.asList(1L, "a"),
Arrays.asList(null, "b"))));
+ assertTrue(error.getMessage(), error.getMessage().contains("'id' index
1"));
+ }
+
+ @Test
+ public void rejectsAValueOfTheWrongType() {
+ IllegalArgumentException error =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> RecordBatches.ofRows(SCHEMA,
List.of(Arrays.asList("1", "a"))));
+ assertTrue(error.getMessage(), error.getMessage().contains("String is
not a INT64"));
+ }
+
+ @Test
+ public void rejectsAnIntegerOutsideItsKind() {
+ Field field = new Field("small", ColumnType.of(ColumnType.Kind.INT8),
false);
+
+ assertThrows(
+ IllegalArgumentException.class, () -> new
ObjectValueVector(field, List.of(128)));
+ assertEquals((byte) -128, new ObjectValueVector(field, List.of((byte)
-128)).getObject(0));
+ }
+}
diff --git a/maven-projects/io-parquet/pom.xml
b/maven-projects/io-parquet/pom.xml
new file mode 100644
index 00000000..ca196039
--- /dev/null
+++ b/maven-projects/io-parquet/pom.xml
@@ -0,0 +1,120 @@
+<?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.4.2</hadoop.version>
+ <parquet.version>1.18.0</parquet.version>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.graphar</groupId>
+ <artifactId>graphar-io-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.graphar</groupId>
+ <artifactId>graphar-storage-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-hadoop</artifactId>
+ <version>${parquet.version}</version>
+ </dependency>
+ <!-- parquet-hadoop needs Hadoop Configuration at runtime; the shaded
client relocates Hadoop's own dependencies. -->
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-api</artifactId>
+ <version>${hadoop.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-runtime</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.13.2</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.graphar</groupId>
+ <artifactId>graphar-storage-local</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>com.diffplug.spotless</groupId>
+ <artifactId>spotless-maven-plugin</artifactId>
+ <version>${spotless-maven-plugin.version}</version>
+ <configuration>
+ <java>
+ <googleJavaFormat>
+ <version>1.7</version>
+ <style>AOSP</style>
+ </googleJavaFormat>
+ </java>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>attach-javadocs</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetBatchCursor.java
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetBatchCursor.java
new file mode 100644
index 00000000..c991d84b
--- /dev/null
+++
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetBatchCursor.java
@@ -0,0 +1,363 @@
+/*
+ * 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 (limit == 0) {
+ 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;
+ closeReader();
+ }
+ return true;
+ }
+ } catch (MissingOffsetIndexException exception) {
+ try {
+ closeReader();
+ } catch (IOException closeException) {
+ exception.addSuppressed(closeException);
+ }
+ throw new UnsupportedOperationException(
+ "A partial row-group range requires a Parquet 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 listGroup = group.getGroup(index, 0);
+ if (listGroup.getType().getFieldCount() != 1) {
+ throw new IllegalArgumentException(
+ "Unsupported Parquet LIST field: " +
column.field().name());
+ }
+ int count = listGroup.getFieldRepetitionCount(0);
+ if (count == 0) {
+ return List.of();
+ }
+ List<Object> values = new ArrayList<>(count);
+ org.apache.graphar.io.ColumnType elementType =
+ column.field().type().elementType().orElseThrow();
+ for (int position = 0; position < count; position++) {
+ Group repeatedGroup = listGroup.getGroup(0, position);
+ if (repeatedGroup.getType().getFieldCount() != 1) {
+ throw new IllegalArgumentException(
+ "Unsupported Parquet LIST element: " +
column.field().name());
+ }
+ values.add(
+ repeatedGroup.getFieldRepetitionCount(0) == 0
+ ? null
+ : scalarValue(repeatedGroup, 0, elementType, 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 =
+ rowGroup.whole
+ ? fileReader.readRowGroup(rowGroup.index)
+ : fileReader.readFilteredRowGroup(
+ rowGroup.index,
+ RowRanges.builder()
+ .addSelectedRange(rowGroup.start,
rowGroup.end - 1)
+ .build());
+ MessageColumnIO columnIO = new
ColumnIOFactory().getColumnIO(readSchema, fileSchema);
+ rows = columnIO.getRecordReader(pages, new
GroupRecordConverter(readSchema));
+ rowsRemainingInGroup = pages.getRowCount();
+ if (rowsRemainingInGroup == 0) {
+ closePages();
+ return openNextRowGroup();
+ }
+ return true;
+ }
+
+ private void closePages() throws IOException {
+ rows = null;
+ rowsRemainingInGroup = 0;
+ if (pages != null) {
+ PageReadStore openPages = pages;
+ pages = null;
+ openPages.close();
+ }
+ }
+
+ private static Map<String, Integer> indexes(List<ParquetColumn> columns) {
+ Map<String, Integer> indexes = new HashMap<>();
+ for (int index = 0; index < columns.size(); index++) {
+ indexes.put(columns.get(index).field().name(), index);
+ }
+ return indexes;
+ }
+
+ private static int[] outputIndexes(
+ List<ParquetColumn> outputColumns, Map<String, Integer>
readColumnIndexes) {
+ int[] indexes = new int[outputColumns.size()];
+ for (int index = 0; index < outputColumns.size(); index++) {
+ indexes[index] =
readColumnIndexes.get(outputColumns.get(index).field().name());
+ }
+ return indexes;
+ }
+
+ private BlockRange nextRange() {
+ while (nextRowGroup < rowGroups.size()) {
+ BlockRange rowGroup = rowGroups.get(nextRowGroup++);
+ long begin = Math.max(rangeStart, rowGroup.start);
+ long end = Math.min(rangeEnd, rowGroup.end);
+ if (begin < end) {
+ return new BlockRange(
+ rowGroup.index,
+ begin - rowGroup.start,
+ end - rowGroup.start,
+ begin == rowGroup.start && end == rowGroup.end);
+ }
+ }
+ return null;
+ }
+
+ private static List<BlockRange> rowGroups(List<BlockMetaData> blocks) {
+ List<BlockRange> result = new ArrayList<>(blocks.size());
+ long start = 0;
+ for (int index = 0; index < blocks.size(); index++) {
+ long end = Math.addExact(start, blocks.get(index).getRowCount());
+ result.add(new BlockRange(index, start, end, true));
+ start = end;
+ }
+ return List.copyOf(result);
+ }
+
+ private static final class BlockRange {
+ private final int index;
+ private final long start;
+ private final long end;
+ private final boolean whole;
+
+ private BlockRange(int index, long start, long end, boolean whole) {
+ this.index = index;
+ this.start = start;
+ this.end = end;
+ this.whole = whole;
+ }
+ }
+}
diff --git
a/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetColumn.java
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetColumn.java
new file mode 100644
index 00000000..9aeedd2c
--- /dev/null
+++
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetColumn.java
@@ -0,0 +1,42 @@
+/*
+ * 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 org.apache.graphar.io.Field;
+import org.apache.parquet.schema.Type;
+
+/** One supported Parquet field and its neutral field contract. */
+final class ParquetColumn {
+ private final Type parquetType;
+ private final Field field;
+
+ ParquetColumn(Type parquetType, Field field) {
+ this.parquetType = parquetType;
+ this.field = field;
+ }
+
+ Type parquetType() {
+ return parquetType;
+ }
+
+ Field field() {
+ return field;
+ }
+}
diff --git
a/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetInputFile.java
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetInputFile.java
new file mode 100644
index 00000000..8edb0473
--- /dev/null
+++
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetInputFile.java
@@ -0,0 +1,44 @@
+/*
+ * 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.InputFile;
+import org.apache.parquet.io.SeekableInputStream;
+
+/** Adapts a GraphAr storage input file to Parquet's input-file interface. */
+final class ParquetInputFile implements org.apache.parquet.io.InputFile {
+ private final InputFile inputFile;
+
+ ParquetInputFile(InputFile inputFile) {
+ this.inputFile = Objects.requireNonNull(inputFile, "inputFile");
+ }
+
+ @Override
+ public long getLength() throws IOException {
+ return inputFile.size();
+ }
+
+ @Override
+ public SeekableInputStream newStream() throws IOException {
+ return new ParquetSeekableInputStream(inputFile.open());
+ }
+}
diff --git
a/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetOutputFile.java
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetOutputFile.java
new file mode 100644
index 00000000..c35de5f3
--- /dev/null
+++
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetOutputFile.java
@@ -0,0 +1,95 @@
+/*
+ * 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 final byte[] single = new byte[1];
+
+ private ParquetPositionOutputStream(PositionOutput output) {
+ this.output = output;
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return output.position();
+ }
+
+ @Override
+ public void write(int value) throws IOException {
+ single[0] = (byte) value;
+ output.write(single);
+ }
+
+ @Override
+ public void write(byte[] bytes, int offset, int length) throws
IOException {
+ output.write(bytes, offset, length);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ output.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ output.close();
+ }
+ }
+}
diff --git
a/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetPhysicalReader.java
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetPhysicalReader.java
new file mode 100644
index 00000000..b9075f56
--- /dev/null
+++
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetPhysicalReader.java
@@ -0,0 +1,433 @@
+/*
+ * 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 final Storage storage;
+ private final FooterCache footers;
+
+ /**
+ * Creates a reader that resolves each request URI through {@code storage}
and parses the footer
+ * of every file it opens.
+ */
+ public ParquetPhysicalReader(Storage storage) {
+ this(storage, 0);
+ }
+
+ /**
+ * Creates a reader that keeps at most {@code footerCacheCapacity} Parquet
footers in memory, so
+ * repeated range reads of one file parse its footer once. A capacity of
zero parses the footer
+ * on every request.
+ *
+ * <p>A remembered footer is keyed by URI and file size only. The caller
must guarantee that no
+ * file read through this reader is rewritten in place while the reader is
in use: a file
+ * replaced with different content of the same size would be read with a
stale footer. Use a
+ * non-zero capacity only for immutable, published datasets.
+ *
+ * @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)) {
+ throw new IllegalArgumentException("Unsupported Parquet LIST
field: " + type);
+ }
+ return ColumnType.listOfElement(
+ new Field(
+ "element",
+ primitiveType(element.asPrimitiveType()),
+ element.isRepetition(Type.Repetition.OPTIONAL)));
+ }
+
+ private static ColumnType primitiveType(PrimitiveType type) {
+ LogicalTypeAnnotation logical = type.getLogicalTypeAnnotation();
+ if (logical instanceof
LogicalTypeAnnotation.DateLogicalTypeAnnotation) {
+ requirePhysical(type, PrimitiveType.PrimitiveTypeName.INT32,
"DATE");
+ return ColumnType.of(ColumnType.Kind.DATE);
+ }
+ if (logical instanceof
LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) {
+ LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestamp =
+ (LogicalTypeAnnotation.TimestampLogicalTypeAnnotation)
logical;
+ requirePhysical(type, PrimitiveType.PrimitiveTypeName.INT64,
"TIMESTAMP");
+ if (timestamp.getUnit() != LogicalTypeAnnotation.TimeUnit.MILLIS) {
+ throw unsupported(type, "only millisecond timestamps are
supported");
+ }
+ return ColumnType.of(ColumnType.Kind.TIMESTAMP_MILLIS);
+ }
+ if (logical instanceof LogicalTypeAnnotation.IntLogicalTypeAnnotation)
{
+ LogicalTypeAnnotation.IntLogicalTypeAnnotation integer =
+ (LogicalTypeAnnotation.IntLogicalTypeAnnotation) logical;
+ if (!integer.isSigned()) {
+ throw unsupported(type, "unsigned integer logical types are
not supported");
+ }
+ switch (integer.getBitWidth()) {
+ case 8:
+ requirePhysical(type,
PrimitiveType.PrimitiveTypeName.INT32, "INT(8)");
+ return ColumnType.of(ColumnType.Kind.INT8);
+ case 16:
+ requirePhysical(type,
PrimitiveType.PrimitiveTypeName.INT32, "INT(16)");
+ return ColumnType.of(ColumnType.Kind.INT16);
+ case 32:
+ requirePhysical(type,
PrimitiveType.PrimitiveTypeName.INT32, "INT(32)");
+ return ColumnType.of(ColumnType.Kind.INT32);
+ case 64:
+ requirePhysical(type,
PrimitiveType.PrimitiveTypeName.INT64, "INT(64)");
+ return ColumnType.of(ColumnType.Kind.INT64);
+ default:
+ throw unsupported(type, "unsupported integer bit width");
+ }
+ }
+ if (logical instanceof
LogicalTypeAnnotation.StringLogicalTypeAnnotation) {
+ if (type.getPrimitiveTypeName() !=
PrimitiveType.PrimitiveTypeName.BINARY
+ && type.getPrimitiveTypeName()
+ !=
PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) {
+ throw unsupported(type, "STRING must use BINARY or
FIXED_LEN_BYTE_ARRAY");
+ }
+ return ColumnType.of(ColumnType.Kind.STRING);
+ }
+ if (logical != null) {
+ throw unsupported(type, "unsupported logical type " + logical);
+ }
+ switch (type.getPrimitiveTypeName()) {
+ case BOOLEAN:
+ return ColumnType.of(ColumnType.Kind.BOOLEAN);
+ case INT32:
+ return ColumnType.of(ColumnType.Kind.INT32);
+ case INT64:
+ return ColumnType.of(ColumnType.Kind.INT64);
+ case FLOAT:
+ return ColumnType.of(ColumnType.Kind.FLOAT32);
+ case DOUBLE:
+ return ColumnType.of(ColumnType.Kind.FLOAT64);
+ case BINARY:
+ case FIXED_LEN_BYTE_ARRAY:
+ return ColumnType.of(ColumnType.Kind.BINARY);
+ default:
+ throw unsupported(type, "unsupported primitive type");
+ }
+ }
+
+ private static void requirePhysical(
+ PrimitiveType type, PrimitiveType.PrimitiveTypeName expected,
String logicalName) {
+ if (type.getPrimitiveTypeName() != expected) {
+ throw unsupported(type, logicalName + " must use " + expected);
+ }
+ }
+
+ private static IllegalArgumentException unsupported(PrimitiveType type,
String reason) {
+ return new IllegalArgumentException(
+ "Unsupported Parquet column " + type.getName() + ": " +
reason);
+ }
+
+ /**
+ * A bounded least-recently-used cache of Parquet footers. A remembered
footer is only reused
+ * when the file still reports the size it had when the footer was parsed.
+ */
+ private static final class FooterCache {
+ private final int capacity;
+ private final LinkedHashMap<URI, Entry> entries;
+
+ private FooterCache(int capacity) {
+ if (capacity < 0) {
+ throw new IllegalArgumentException(
+ "Parquet footer cache capacity cannot be negative: " +
capacity);
+ }
+ this.capacity = capacity;
+ this.entries =
+ new LinkedHashMap<>(16, 0.75f, true) {
+ @Override
+ protected boolean removeEldestEntry(Map.Entry<URI,
Entry> eldest) {
+ return size() > FooterCache.this.capacity;
+ }
+ };
+ }
+
+ private synchronized ParquetMetadata get(URI uri, long size) {
+ if (capacity == 0) {
+ return null;
+ }
+ Entry entry = entries.get(uri);
+ if (entry == null || entry.size != size) {
+ return null;
+ }
+ return entry.footer;
+ }
+
+ private synchronized void put(URI uri, long size, ParquetMetadata
footer) {
+ if (capacity == 0) {
+ return;
+ }
+ entries.put(uri, new Entry(size, footer));
+ }
+
+ private static final class Entry {
+ private final long size;
+ private final ParquetMetadata footer;
+
+ private Entry(long size, ParquetMetadata footer) {
+ this.size = size;
+ this.footer = footer;
+ }
+ }
+ }
+}
diff --git
a/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetPhysicalWriter.java
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetPhysicalWriter.java
new file mode 100644
index 00000000..95b6773b
--- /dev/null
+++
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetPhysicalWriter.java
@@ -0,0 +1,291 @@
+/*
+ * 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) {
+ Field element = field.type().children().get(0);
+ return (field.nullable() ? Types.optionalList() : Types.requiredList())
+ .element(listElementType(element))
+ .named(field.name());
+ }
+
+ private static Type listElementType(Field element) {
+ Types.PrimitiveBuilder<PrimitiveType> builder =
+ Types.primitive(physicalType(element.type()),
repetition(element));
+ LogicalTypeAnnotation logical = logicalType(element.type());
+ if (logical != null) {
+ builder.as(logical);
+ }
+ return builder.named("element");
+ }
+
+ private static PrimitiveType.PrimitiveTypeName physicalType(ColumnType
type) {
+ switch (type.kind()) {
+ case BOOLEAN:
+ return PrimitiveType.PrimitiveTypeName.BOOLEAN;
+ case INT8:
+ case INT16:
+ case INT32:
+ case DATE:
+ return PrimitiveType.PrimitiveTypeName.INT32;
+ case INT64:
+ case TIMESTAMP_MILLIS:
+ return PrimitiveType.PrimitiveTypeName.INT64;
+ case FLOAT32:
+ return PrimitiveType.PrimitiveTypeName.FLOAT;
+ case FLOAT64:
+ return PrimitiveType.PrimitiveTypeName.DOUBLE;
+ case STRING:
+ case BINARY:
+ return PrimitiveType.PrimitiveTypeName.BINARY;
+ case LIST:
+ default:
+ throw new IllegalArgumentException("Unsupported flat Parquet
type: " + type.kind());
+ }
+ }
+
+ private static Type.Repetition repetition(Field field) {
+ return field.nullable() ? Type.Repetition.OPTIONAL :
Type.Repetition.REQUIRED;
+ }
+
+ private static LogicalTypeAnnotation logicalType(ColumnType type) {
+ switch (type.kind()) {
+ case INT8:
+ return LogicalTypeAnnotation.intType(8, true);
+ case INT16:
+ return LogicalTypeAnnotation.intType(16, true);
+ case DATE:
+ return LogicalTypeAnnotation.dateType();
+ case TIMESTAMP_MILLIS:
+ return LogicalTypeAnnotation.timestampType(
+ true, LogicalTypeAnnotation.TimeUnit.MILLIS);
+ case STRING:
+ return LogicalTypeAnnotation.stringType();
+ default:
+ return null;
+ }
+ }
+
+ private static void requireSchema(Schema expected, Schema actual) {
+ List<Field> expectedFields = expected.fields();
+ List<Field> actualFields = actual.fields();
+ if (expectedFields.size() != actualFields.size()) {
+ throw new IllegalArgumentException("Record batch schema does not
match write request.");
+ }
+ for (int index = 0; index < expectedFields.size(); index++) {
+ Field left = expectedFields.get(index);
+ Field right = actualFields.get(index);
+ if (!left.name().equals(right.name())
+ || !left.type().equals(right.type())
+ || left.nullable() != right.nullable()) {
+ throw new IllegalArgumentException(
+ "Record batch schema does not match write request.");
+ }
+ }
+ }
+
+ private static Group toGroup(
+ SimpleGroupFactory groups, Schema schema, RecordBatch batch, int
rowIndex) {
+ Group group = groups.newGroup();
+ for (int index = 0; index < schema.fields().size(); index++) {
+ Field field = schema.fields().get(index);
+ Object value = batch.column(index).getObject(rowIndex);
+ if (value == null) {
+ if (!field.nullable()) {
+ throw new IllegalArgumentException("Required field is
null: " + field.name());
+ }
+ continue;
+ }
+ add(group, field, value);
+ }
+ return group;
+ }
+
+ private static void add(Group group, Field field, Object value) {
+ String name = field.name();
+ switch (field.type().kind()) {
+ case BOOLEAN:
+ group.add(name, require(value, Boolean.class, name));
+ return;
+ case INT8:
+ case INT16:
+ case INT32:
+ group.add(name, require(value, Number.class, name).intValue());
+ return;
+ case INT64:
+ group.add(name, require(value, Number.class,
name).longValue());
+ return;
+ case FLOAT32:
+ group.add(name, require(value, Number.class,
name).floatValue());
+ return;
+ case FLOAT64:
+ group.add(name, require(value, Number.class,
name).doubleValue());
+ return;
+ case STRING:
+ group.add(name, Binary.fromString(require(value, String.class,
name)));
+ return;
+ case BINARY:
+ ByteBuffer bytes = require(value, ByteBuffer.class,
name).asReadOnlyBuffer();
+ byte[] copy = new byte[bytes.remaining()];
+ bytes.get(copy);
+ group.add(name, Binary.fromConstantByteArray(copy));
+ return;
+ case DATE:
+ group.add(
+ name, Math.toIntExact(require(value, LocalDate.class,
name).toEpochDay()));
+ return;
+ case TIMESTAMP_MILLIS:
+ group.add(name, require(value, Instant.class,
name).toEpochMilli());
+ return;
+ case LIST:
+ addList(group, field, value);
+ return;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported Parquet type: " + field.type().kind());
+ }
+ }
+
+ private static void addList(Group group, Field field, Object value) {
+ if (!(value instanceof List<?>)) {
+ throw new IllegalArgumentException(
+ "Unexpected value for " + field.name() + ": expected
List");
+ }
+ Group list = group.addGroup(field.name());
+ Field element = field.type().children().get(0);
+ for (Object elementValue : (List<?>) value) {
+ Group repeated = list.addGroup("list");
+ if (elementValue != null) {
+ add(repeated, element, elementValue);
+ } else if (!element.nullable()) {
+ throw new IllegalArgumentException(
+ "Required LIST element is null: " + field.name());
+ }
+ }
+ }
+
+ private static <T> T require(Object value, Class<T> type, String name) {
+ if (!type.isInstance(value)) {
+ throw new IllegalArgumentException(
+ "Unexpected value for " + name + ": expected " +
type.getSimpleName());
+ }
+ return type.cast(value);
+ }
+}
diff --git
a/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetSeekableInputStream.java
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetSeekableInputStream.java
new file mode 100644
index 00000000..2a2a9d77
--- /dev/null
+++
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetSeekableInputStream.java
@@ -0,0 +1,88 @@
+/*
+ * 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.util.Objects;
+import org.apache.graphar.storage.SeekableInput;
+
+/** Adapts a GraphAr seekable input to Parquet's seekable input stream. */
+final class ParquetSeekableInputStream extends
org.apache.parquet.io.SeekableInputStream {
+ private final SeekableInput input;
+
+ ParquetSeekableInputStream(SeekableInput input) {
+ this.input = Objects.requireNonNull(input, "input");
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return input.position();
+ }
+
+ @Override
+ public void seek(long newPos) throws IOException {
+ input.seek(newPos);
+ }
+
+ @Override
+ public int read() throws IOException {
+ byte[] singleByte = new byte[1];
+ return read(singleByte, 0, 1) < 0 ? -1 :
Byte.toUnsignedInt(singleByte[0]);
+ }
+
+ @Override
+ public int read(byte[] bytes, int offset, int length) throws IOException {
+ Objects.requireNonNull(bytes, "bytes");
+ Objects.checkFromIndexSize(offset, length, bytes.length);
+ if (length == 0) {
+ return 0;
+ }
+ return input.read(ByteBuffer.wrap(bytes, offset, length));
+ }
+
+ @Override
+ public void readFully(byte[] bytes) throws IOException {
+ Objects.requireNonNull(bytes, "bytes");
+ input.readFully(ByteBuffer.wrap(bytes));
+ }
+
+ @Override
+ public void readFully(byte[] bytes, int offset, int length) throws
IOException {
+ Objects.requireNonNull(bytes, "bytes");
+ Objects.checkFromIndexSize(offset, length, bytes.length);
+ input.readFully(ByteBuffer.wrap(bytes, offset, length));
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws IOException {
+ return input.read(buffer);
+ }
+
+ @Override
+ public void readFully(ByteBuffer buffer) throws IOException {
+ input.readFully(buffer);
+ }
+
+ @Override
+ public void close() throws IOException {
+ input.close();
+ }
+}
diff --git
a/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetValueVector.java
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetValueVector.java
new file mode 100644
index 00000000..c4b2fa48
--- /dev/null
+++
b/maven-projects/io-parquet/src/main/java/org/apache/graphar/io/parquet/ParquetValueVector.java
@@ -0,0 +1,58 @@
+/*
+ * 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.nio.ByteBuffer;
+import org.apache.graphar.io.Field;
+import org.apache.graphar.io.ValueVector;
+
+/** One column of values materialized from a Parquet row group, with defensive
binary views. */
+final class ParquetValueVector implements ValueVector {
+ private final Field field;
+ private final Object[] values;
+
+ ParquetValueVector(Field field, Object[] values) {
+ this.field = field;
+ this.values = values;
+ }
+
+ @Override
+ public Field field() {
+ return field;
+ }
+
+ @Override
+ public int valueCount() {
+ return values.length;
+ }
+
+ @Override
+ public boolean isNull(int index) {
+ return values[index] == null;
+ }
+
+ @Override
+ public Object getObject(int index) {
+ Object value = values[index];
+ if (value instanceof byte[]) {
+ return ByteBuffer.wrap((byte[]) value).asReadOnlyBuffer();
+ }
+ return value;
+ }
+}
diff --git
a/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetListRoundTripTest.java
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetListRoundTripTest.java
new file mode 100644
index 00000000..c628dcb8
--- /dev/null
+++
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetListRoundTripTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.graphar.io.BatchCursor;
+import org.apache.graphar.io.ColumnType;
+import org.apache.graphar.io.Field;
+import org.apache.graphar.io.RecordBatch;
+import org.apache.graphar.io.RecordBatches;
+import org.apache.graphar.io.Schema;
+import org.apache.graphar.io.WriteMode;
+import org.apache.graphar.io.WriteRequest;
+import org.apache.graphar.storage.local.LocalStorage;
+import org.junit.Test;
+
+/**
+ * Verifies standard Parquet LIST values, empty lists, and null lists
round-trip through neutral IO.
+ */
+public class ParquetListRoundTripTest {
+ private static final Schema SCHEMA =
+ new Schema(
+ List.of(
+ new Field("id",
ColumnType.of(ColumnType.Kind.INT64), false),
+ new Field(
+ "tags",
+
ColumnType.listOf(ColumnType.of(ColumnType.Kind.STRING)),
+ true)));
+
+ @Test
+ public void roundTripsNullableStringLists() throws Exception {
+ Path file = Files.createTempFile("graphar-list-", ".parquet");
+ Files.deleteIfExists(file);
+ LocalStorage storage = new LocalStorage();
+ URI uri = file.toUri();
+ try {
+ new ParquetPhysicalWriter(storage)
+ .write(
+ new WriteRequest(uri, SCHEMA,
WriteMode.CREATE_NEW),
+ new OneBatch(
+ RecordBatches.ofRows(
+ SCHEMA,
+ List.of(
+ Arrays.asList(1L,
List.of("one", "two")),
+ Arrays.asList(2L,
List.of()),
+ Arrays.asList(3L,
null)))));
+ List<Object> lists = new ArrayList<>();
+ try (BatchCursor cursor =
+ new ParquetPhysicalReader(storage)
+
.read(org.apache.graphar.io.ReadRequest.builder(uri).build())
+ .cursor()) {
+ while (cursor.next()) {
+ RecordBatch batch = cursor.batch();
+ for (int index = 0; index < batch.rowCount(); index++) {
+ lists.add(batch.column(1).getObject(index));
+ }
+ }
+ }
+ assertEquals(List.of("one", "two"), lists.get(0));
+ assertEquals(List.of(), lists.get(1));
+ assertEquals(null, lists.get(2));
+ } finally {
+ Files.deleteIfExists(file);
+ }
+ }
+
+ private static final class OneBatch implements BatchCursor {
+ private final RecordBatch batch;
+ private boolean advanced;
+
+ private OneBatch(RecordBatch batch) {
+ this.batch = batch;
+ }
+
+ @Override
+ public boolean next() {
+ if (advanced) {
+ return false;
+ }
+ advanced = true;
+ return true;
+ }
+
+ @Override
+ public RecordBatch batch() {
+ return batch;
+ }
+
+ @Override
+ public void close() {}
+ }
+}
diff --git
a/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalIoEfficiencyTest.java
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalIoEfficiencyTest.java
new file mode 100644
index 00000000..0c6c92be
--- /dev/null
+++
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalIoEfficiencyTest.java
@@ -0,0 +1,483 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import org.apache.graphar.io.BatchCursor;
+import org.apache.graphar.io.ColumnRef;
+import org.apache.graphar.io.ColumnType;
+import org.apache.graphar.io.Field;
+import org.apache.graphar.io.Projection;
+import org.apache.graphar.io.ReadRequest;
+import org.apache.graphar.io.ReadResult;
+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.WriteMode;
+import org.apache.graphar.io.WriteRequest;
+import org.apache.graphar.storage.InputFile;
+import org.apache.graphar.storage.OutputFile;
+import org.apache.graphar.storage.PositionOutput;
+import org.apache.graphar.storage.SeekableInput;
+import org.apache.graphar.storage.Storage;
+import org.apache.graphar.storage.local.LocalStorage;
+import org.junit.Test;
+
+/** Regression gate for physical I/O of an indexed, projected Parquet range
read. */
+public class ParquetPhysicalIoEfficiencyTest {
+ private static final int ROW_COUNT = 262_144;
+ private static final int BATCH_ROWS = 1_024;
+ private static final long RANGE_START = 131_456;
+ private static final int RANGE_ROWS = 32;
+ private static final Schema SCHEMA =
+ new Schema(
+ List.of(
+ new Field("id",
ColumnType.of(ColumnType.Kind.INT64), false),
+ new Field("payload",
ColumnType.of(ColumnType.Kind.INT64), false)));
+
+ @Test
+ public void readsAnIndexedProjectedRangeWithoutReadingTheWholeFile()
throws Exception {
+ Path directory =
Files.createTempDirectory("graphar-parquet-io-efficiency-");
+ Path file = directory.resolve("topology.parquet");
+ CountingStorage storage = new CountingStorage(new LocalStorage());
+ URI uri = file.toUri();
+ try {
+ new ParquetPhysicalWriter(storage)
+ .write(
+ new WriteRequest(uri, SCHEMA,
WriteMode.CREATE_NEW),
+ new GeneratedBatches(ROW_COUNT, BATCH_ROWS));
+
+ long fileBytes = Files.size(file);
+ Counters write = storage.outputCounters();
+ storage.resetInputCounters();
+
+ ReadResult result =
+ new ParquetPhysicalReader(storage)
+ .read(
+ ReadRequest.builder(uri)
+
.projection(Projection.of(ColumnRef.of("payload")))
+ .rowRange(
+ new RowRange(
+ RANGE_START,
RANGE_START + RANGE_ROWS))
+ .build());
+ long rows = assertPayloads(result);
+ Counters read = storage.inputCounters();
+
+ System.out.println(
+ "PARQUET_IO_BASELINE rows="
+ + ROW_COUNT
+ + " file_bytes="
+ + fileBytes
+ + " write_bytes="
+ + write.bytes
+ + " write_calls="
+ + write.calls
+ + " read_bytes="
+ + read.bytes
+ + " read_calls="
+ + read.calls
+ + " read_seeks="
+ + read.seeks
+ + " read_opens="
+ + read.opens);
+ assertEquals(RANGE_ROWS, rows);
+ assertEquals(1, read.opens);
+ assertTrue(
+ "The generated data must exercise a materially sized
file.",
+ fileBytes > 1_000_000);
+ assertTrue(
+ "A projected 32-row range must not read even one eighth of
the Parquet file: "
+ + read.bytes
+ + " of "
+ + fileBytes,
+ read.bytes * 8L < fileBytes);
+ } finally {
+ Files.deleteIfExists(file);
+ Files.deleteIfExists(directory);
+ }
+ }
+
+ @Test
+ public void reusesARememberedFooterAcrossRepeatedReadsOfOneFile() throws
Exception {
+ Path directory =
Files.createTempDirectory("graphar-parquet-footer-cache-");
+ Path file = directory.resolve("topology.parquet");
+ CountingStorage storage = new CountingStorage(new LocalStorage());
+ URI uri = file.toUri();
+ try {
+ new ParquetPhysicalWriter(storage)
+ .write(
+ new WriteRequest(uri, SCHEMA,
WriteMode.CREATE_NEW),
+ new GeneratedBatches(ROW_COUNT, BATCH_ROWS));
+
+ ParquetPhysicalReader uncached = new
ParquetPhysicalReader(storage, 0);
+ assertEquals(RANGE_ROWS, readRange(uncached, uri));
+ storage.resetInputCounters();
+ assertEquals(RANGE_ROWS, readRange(uncached, uri));
+ Counters repeatedWithoutCache = storage.inputCounters();
+
+ ParquetPhysicalReader cached = new ParquetPhysicalReader(storage,
16);
+ assertEquals(RANGE_ROWS, readRange(cached, uri));
+ storage.resetInputCounters();
+ assertEquals(RANGE_ROWS, readRange(cached, uri));
+ Counters repeatedWithCache = storage.inputCounters();
+
+ System.out.println(
+ "PARQUET_FOOTER_CACHE repeat_bytes_uncached="
+ + repeatedWithoutCache.bytes
+ + " repeat_calls_uncached="
+ + repeatedWithoutCache.calls
+ + " repeat_bytes_cached="
+ + repeatedWithCache.bytes
+ + " repeat_calls_cached="
+ + repeatedWithCache.calls);
+ assertTrue(
+ "A remembered footer must remove read work from the
repeated read: "
+ + repeatedWithCache.bytes
+ + " of "
+ + repeatedWithoutCache.bytes,
+ repeatedWithCache.bytes < repeatedWithoutCache.bytes);
+ assertTrue(
+ "A remembered footer must not add stream opens.",
+ repeatedWithCache.opens == repeatedWithoutCache.opens);
+ } finally {
+ Files.deleteIfExists(file);
+ Files.deleteIfExists(directory);
+ }
+ }
+
+ private static long readRange(ParquetPhysicalReader reader, URI uri)
throws IOException {
+ return assertPayloads(
+ reader.read(
+ ReadRequest.builder(uri)
+
.projection(Projection.of(ColumnRef.of("payload")))
+ .rowRange(new RowRange(RANGE_START,
RANGE_START + RANGE_ROWS))
+ .build()));
+ }
+
+ private static long assertPayloads(ReadResult result) throws IOException {
+ long rows = 0;
+ try (BatchCursor cursor = result.cursor()) {
+ while (cursor.next()) {
+ RecordBatch batch = cursor.batch();
+ for (int index = 0; index < batch.rowCount(); index++) {
+ assertEquals(mix(RANGE_START + rows),
batch.column(0).getObject(index));
+ rows++;
+ }
+ }
+ }
+ return rows;
+ }
+
+ private static long mix(long value) {
+ long result = value + 0x9E3779B97F4A7C15L;
+ result = (result ^ (result >>> 30)) * 0xBF58476D1CE4E5B9L;
+ result = (result ^ (result >>> 27)) * 0x94D049BB133111EBL;
+ return result ^ (result >>> 31);
+ }
+
+ private static final class GeneratedBatches implements BatchCursor {
+ private final int rowCount;
+ private final int batchRows;
+ private int nextStart;
+ private RecordBatch current;
+
+ private GeneratedBatches(int rowCount, int batchRows) {
+ this.rowCount = rowCount;
+ this.batchRows = batchRows;
+ }
+
+ @Override
+ public boolean next() {
+ if (nextStart == rowCount) {
+ current = null;
+ return false;
+ }
+ int start = nextStart;
+ int size = Math.min(batchRows, rowCount - start);
+ nextStart += size;
+ current = new GeneratedBatch(start, size);
+ return true;
+ }
+
+ @Override
+ public RecordBatch batch() {
+ if (current == null) {
+ throw new IllegalStateException("No current batch. Call next()
before batch().");
+ }
+ return current;
+ }
+
+ @Override
+ public void close() {}
+ }
+
+ private static final class GeneratedBatch implements RecordBatch {
+ private final int start;
+ private final int rowCount;
+
+ private GeneratedBatch(int start, int rowCount) {
+ this.start = start;
+ this.rowCount = rowCount;
+ }
+
+ @Override
+ public Schema schema() {
+ return SCHEMA;
+ }
+
+ @Override
+ public int rowCount() {
+ return rowCount;
+ }
+
+ @Override
+ public int columnCount() {
+ return SCHEMA.fields().size();
+ }
+
+ @Override
+ public ValueVector column(int columnIndex) {
+ Field field = SCHEMA.fields().get(columnIndex);
+ return new ValueVector() {
+ @Override
+ public Field field() {
+ return field;
+ }
+
+ @Override
+ public int valueCount() {
+ return rowCount;
+ }
+
+ @Override
+ public boolean isNull(int index) {
+ return false;
+ }
+
+ @Override
+ public Object getObject(int index) {
+ if (index < 0 || index >= rowCount) {
+ throw new IndexOutOfBoundsException("Row index: " +
index);
+ }
+ long value = start + index;
+ return columnIndex == 0 ? value : mix(value);
+ }
+ };
+ }
+ }
+
+ private static final class CountingStorage implements Storage {
+ private final Storage delegate;
+ private final Counters input = new Counters();
+ private final Counters output = new Counters();
+
+ private CountingStorage(Storage delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public InputFile inputFile(URI uri) {
+ return new CountingInputFile(delegate.inputFile(uri), input);
+ }
+
+ @Override
+ public OutputFile outputFile(URI uri) {
+ return new CountingOutputFile(delegate.outputFile(uri), output);
+ }
+
+ @Override
+ public boolean exists(URI uri) throws IOException {
+ return delegate.exists(uri);
+ }
+
+ private Counters inputCounters() {
+ return input.copy();
+ }
+
+ private Counters outputCounters() {
+ return output.copy();
+ }
+
+ private void resetInputCounters() {
+ input.reset();
+ }
+ }
+
+ private static final class CountingInputFile implements InputFile {
+ private final InputFile delegate;
+ private final Counters counters;
+
+ private CountingInputFile(InputFile delegate, Counters counters) {
+ this.delegate = delegate;
+ this.counters = counters;
+ }
+
+ @Override
+ public URI uri() {
+ return delegate.uri();
+ }
+
+ @Override
+ public long size() throws IOException {
+ return delegate.size();
+ }
+
+ @Override
+ public SeekableInput open() throws IOException {
+ counters.opens++;
+ return new CountingSeekableInput(delegate.open(), counters);
+ }
+ }
+
+ private static final class CountingOutputFile implements OutputFile {
+ private final OutputFile delegate;
+ private final Counters counters;
+
+ private CountingOutputFile(OutputFile delegate, Counters counters) {
+ this.delegate = delegate;
+ this.counters = counters;
+ }
+
+ @Override
+ public URI uri() {
+ return delegate.uri();
+ }
+
+ @Override
+ public PositionOutput create() throws IOException {
+ counters.opens++;
+ return new CountingPositionOutput(delegate.create(), counters);
+ }
+
+ @Override
+ public PositionOutput createOrOverwrite() throws IOException {
+ counters.opens++;
+ return new CountingPositionOutput(delegate.createOrOverwrite(),
counters);
+ }
+ }
+
+ private static final class CountingSeekableInput implements SeekableInput {
+ private final SeekableInput delegate;
+ private final Counters counters;
+
+ private CountingSeekableInput(SeekableInput delegate, Counters
counters) {
+ this.delegate = delegate;
+ this.counters = counters;
+ }
+
+ @Override
+ public long position() throws IOException {
+ return delegate.position();
+ }
+
+ @Override
+ public void seek(long newPosition) throws IOException {
+ counters.seeks++;
+ delegate.seek(newPosition);
+ }
+
+ @Override
+ public int read(ByteBuffer destination) throws IOException {
+ int read = delegate.read(destination);
+ counters.calls++;
+ if (read > 0) {
+ counters.bytes += read;
+ }
+ return read;
+ }
+
+ @Override
+ public void close() throws IOException {
+ delegate.close();
+ }
+ }
+
+ private static final class CountingPositionOutput implements
PositionOutput {
+ private final PositionOutput delegate;
+ private final Counters counters;
+
+ private CountingPositionOutput(PositionOutput delegate, Counters
counters) {
+ this.delegate = delegate;
+ this.counters = counters;
+ }
+
+ @Override
+ public long position() throws IOException {
+ return delegate.position();
+ }
+
+ @Override
+ public void write(ByteBuffer source) throws IOException {
+ int bytes = source.remaining();
+ delegate.write(source);
+ counters.calls++;
+ counters.bytes += bytes;
+ }
+
+ @Override
+ public void write(byte[] source, int offset, int length) throws
IOException {
+ delegate.write(source, offset, length);
+ counters.calls++;
+ counters.bytes += length;
+ }
+
+ @Override
+ public void flush() throws IOException {
+ delegate.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ delegate.close();
+ }
+ }
+
+ private static final class Counters {
+ private long bytes;
+ private long calls;
+ private long seeks;
+ private long opens;
+
+ private Counters copy() {
+ Counters result = new Counters();
+ result.bytes = bytes;
+ result.calls = calls;
+ result.seeks = seeks;
+ result.opens = opens;
+ return result;
+ }
+
+ private void reset() {
+ bytes = 0;
+ calls = 0;
+ seeks = 0;
+ opens = 0;
+ }
+ }
+}
diff --git
a/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalReaderFixtureTest.java
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalReaderFixtureTest.java
new file mode 100644
index 00000000..a37c519e
--- /dev/null
+++
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalReaderFixtureTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+import org.apache.graphar.io.BatchCursor;
+import org.apache.graphar.io.ColumnRef;
+import org.apache.graphar.io.Projection;
+import org.apache.graphar.io.ReadCapability;
+import org.apache.graphar.io.ReadRequest;
+import org.apache.graphar.io.ReadResult;
+import org.apache.graphar.io.RecordBatch;
+import org.apache.graphar.io.RowRange;
+import org.apache.graphar.storage.local.LocalStorage;
+import org.junit.Test;
+
+public class ParquetPhysicalReaderFixtureTest {
+ @Test
+ public void physicallyAppliesProjectedRangeFromLdbcParquetFixture() throws
IOException {
+ Path fixture =
+ Path.of(
+ "..",
+ "..",
+ "testing",
+ "ldbc_sample",
+ "parquet",
+ "vertex",
+ "person",
+ "firstName_lastName_gender",
+ "chunk0");
+ ReadRequest request =
+ ReadRequest.builder(fixture.toUri())
+ .projection(Projection.of(ColumnRef.of("firstName")))
+ .rowRange(new RowRange(1, 12))
+ .build();
+
+ ReadResult result = new ParquetPhysicalReader(new
LocalStorage()).read(request);
+
+ assertEquals(
+ EnumSet.of(ReadCapability.PROJECTION,
ReadCapability.ROW_RANGE),
+ result.report().applied());
+ assertEquals(EnumSet.noneOf(ReadCapability.class),
result.report().declined());
+ assertEquals(
+ List.of(
+ "Eli", "Joseph", "Yacine", "Jose", "Steve", "John",
"Jun", "A. C.", "Karim",
+ "Hermann", "Lin"),
+ firstNames(result));
+ }
+
+ @Test
+ public void rejectsFiltersUntilTheyCanBeAppliedPhysically() {
+ Path fixture =
+ Path.of(
+ "..",
+ "..",
+ "testing",
+ "ldbc_sample",
+ "parquet",
+ "vertex",
+ "person",
+ "firstName_lastName_gender",
+ "chunk0");
+ ReadRequest request =
+ ReadRequest.builder(fixture.toUri())
+ .filters(
+ List.of(
+ org.apache.graphar.io.Filter.equal(
+ ColumnRef.of("gender"),
+
org.apache.graphar.io.Literal.of("male"))))
+ .build();
+
+ UnsupportedOperationException error =
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> new ParquetPhysicalReader(new
LocalStorage()).read(request));
+ assertEquals(
+ "Parquet filter pushdown is not implemented; refusing semantic
fallback.",
+ error.getMessage());
+ }
+
+ private static List<String> firstNames(ReadResult result) throws
IOException {
+ List<String> firstNames = new ArrayList<>();
+ try (BatchCursor cursor = result.cursor()) {
+ while (cursor.next()) {
+ RecordBatch batch = cursor.batch();
+ assertEquals(1, batch.schema().fields().size());
+ assertEquals("firstName",
batch.schema().fields().get(0).name());
+ for (int index = 0; index < batch.rowCount(); index++) {
+ firstNames.add((String) batch.column(0).getObject(index));
+ }
+ }
+ assertFalse(cursor.next());
+ }
+ return firstNames;
+ }
+}
diff --git
a/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalWriterModeTest.java
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalWriterModeTest.java
new file mode 100644
index 00000000..1c139f9b
--- /dev/null
+++
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalWriterModeTest.java
@@ -0,0 +1,122 @@
+/*
+ * 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 static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import org.apache.graphar.io.BatchCursor;
+import org.apache.graphar.io.ColumnType;
+import org.apache.graphar.io.Field;
+import org.apache.graphar.io.ReadRequest;
+import org.apache.graphar.io.RecordBatches;
+import org.apache.graphar.io.Schema;
+import org.apache.graphar.io.WriteMode;
+import org.apache.graphar.io.WriteRequest;
+import org.apache.graphar.storage.local.LocalStorage;
+import org.junit.Test;
+
+public class ParquetPhysicalWriterModeTest {
+ private static final Schema SCHEMA =
+ new Schema(List.of(new Field("id",
ColumnType.of(ColumnType.Kind.INT64), false)));
+
+ @Test
+ public void rejectsAppendWithoutTouchingTheExistingFile() throws Exception
{
+ Path file = Files.createTempFile("graphar-mode-", ".parquet");
+ Files.deleteIfExists(file);
+ LocalStorage storage = new LocalStorage();
+ URI uri = file.toUri();
+ try {
+ ParquetPhysicalWriter writer = new ParquetPhysicalWriter(storage);
+ writer.write(new WriteRequest(uri, SCHEMA, WriteMode.CREATE_NEW),
batch(7L));
+ byte[] before = Files.readAllBytes(file);
+ UnsupportedOperationException failure =
+ assertThrows(
+ UnsupportedOperationException.class,
+ () ->
+ writer.write(
+ new WriteRequest(uri, SCHEMA,
WriteMode.APPEND),
+ batch(8L)));
+ assertTrue(failure.getMessage(),
failure.getMessage().contains("APPEND"));
+ assertArrayEquals(before, Files.readAllBytes(file));
+ assertEquals(List.of(7L), ids(storage, uri));
+ } finally {
+ Files.deleteIfExists(file);
+ }
+ }
+
+ @Test
+ public void overwriteReplacesTheExistingFile() throws Exception {
+ Path file = Files.createTempFile("graphar-mode-", ".parquet");
+ Files.deleteIfExists(file);
+ LocalStorage storage = new LocalStorage();
+ URI uri = file.toUri();
+ try {
+ ParquetPhysicalWriter writer = new ParquetPhysicalWriter(storage);
+ writer.write(new WriteRequest(uri, SCHEMA, WriteMode.CREATE_NEW),
batch(7L));
+ writer.write(new WriteRequest(uri, SCHEMA, WriteMode.OVERWRITE),
batch(8L));
+ assertEquals(List.of(8L), ids(storage, uri));
+ } finally {
+ Files.deleteIfExists(file);
+ }
+ }
+
+ private static BatchCursor batch(long id) {
+ return new BatchCursor() {
+ private boolean available = true;
+
+ @Override
+ public boolean next() {
+ boolean result = available;
+ available = false;
+ return result;
+ }
+
+ @Override
+ public org.apache.graphar.io.RecordBatch batch() {
+ return RecordBatches.ofArrays(SCHEMA, List.<Object[]>of(new
Object[] {id}));
+ }
+
+ @Override
+ public void close() {}
+ };
+ }
+
+ private static List<Object> ids(LocalStorage storage, URI uri) throws
Exception {
+ List<Object> ids = new java.util.ArrayList<>();
+ try (BatchCursor cursor =
+ new ParquetPhysicalReader(storage)
+ .read(ReadRequest.builder(uri).build())
+ .cursor()) {
+ while (cursor.next()) {
+ for (int index = 0; index < cursor.batch().rowCount();
index++) {
+ ids.add(cursor.batch().column(0).getObject(index));
+ }
+ }
+ }
+ return ids;
+ }
+}
diff --git
a/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalWriterOffsetIndexTest.java
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalWriterOffsetIndexTest.java
new file mode 100644
index 00000000..1a285214
--- /dev/null
+++
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetPhysicalWriterOffsetIndexTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+import org.apache.graphar.io.BatchCursor;
+import org.apache.graphar.io.ColumnRef;
+import org.apache.graphar.io.ColumnType;
+import org.apache.graphar.io.Field;
+import org.apache.graphar.io.Projection;
+import org.apache.graphar.io.ReadCapability;
+import org.apache.graphar.io.ReadRequest;
+import org.apache.graphar.io.ReadResult;
+import org.apache.graphar.io.RecordBatch;
+import org.apache.graphar.io.RecordBatches;
+import org.apache.graphar.io.RowRange;
+import org.apache.graphar.io.Schema;
+import org.apache.graphar.io.WriteMode;
+import org.apache.graphar.io.WriteRequest;
+import org.apache.graphar.storage.local.LocalStorage;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.internal.column.columnindex.OffsetIndex;
+import org.junit.Test;
+
+public class ParquetPhysicalWriterOffsetIndexTest {
+ private static final Schema TOPOLOGY_SCHEMA =
+ new Schema(
+ List.of(
+ new Field(
+ "_graphArSrcIndex",
+ ColumnType.of(ColumnType.Kind.INT64),
+ false),
+ new Field(
+ "_graphArDstIndex",
+ ColumnType.of(ColumnType.Kind.INT64),
+ false)));
+
+ @Test
+ public void writesOffsetIndexesForPhysicalTopologyRangeReads() throws
Exception {
+ Path directory =
Files.createTempDirectory("graphar-parquet-offset-index-");
+ Path file = directory.resolve("topology.parquet");
+ LocalStorage storage = new LocalStorage();
+ URI uri = file.toUri();
+ try {
+ new ParquetPhysicalWriter(storage)
+ .write(
+ new WriteRequest(uri, TOPOLOGY_SCHEMA,
WriteMode.CREATE_NEW),
+ topologyBatches());
+
+ assertOffsetIndexes(storage, uri);
+
+ ReadResult result =
+ new ParquetPhysicalReader(storage)
+ .read(
+ ReadRequest.builder(uri)
+ .projection(
+
Projection.of(ColumnRef.of("_graphArDstIndex")))
+ .rowRange(new RowRange(2051, 2063))
+ .build());
+ assertEquals(
+ EnumSet.of(ReadCapability.PROJECTION,
ReadCapability.ROW_RANGE),
+ result.report().applied());
+ assertTrue(result.report().declined().isEmpty());
+ assertEquals(
+ List.of(
+ 12051L, 12052L, 12053L, 12054L, 12055L, 12056L,
12057L, 12058L, 12059L,
+ 12060L, 12061L, 12062L),
+ destinations(result));
+ } finally {
+ Files.deleteIfExists(file);
+ Files.deleteIfExists(directory);
+ }
+ }
+
+ private static void assertOffsetIndexes(LocalStorage storage, URI uri)
throws IOException {
+ try (ParquetFileReader fileReader =
+ ParquetFileReader.open(new
ParquetInputFile(storage.inputFile(uri)))) {
+ assertEquals(1, fileReader.getRowGroups().size());
+ BlockMetaData rowGroup = fileReader.getRowGroups().get(0);
+ for (ColumnChunkMetaData column : rowGroup.getColumns()) {
+ OffsetIndex offsetIndex = fileReader.readOffsetIndex(column);
+ assertNotNull("Missing Offset Index for " + column.getPath(),
offsetIndex);
+ assertEquals(
+ "Expected bounded pages for " + column.getPath(),
+ 4,
+ offsetIndex.getPageCount());
+ }
+ }
+ }
+
+ private static BatchCursor topologyBatches() {
+ List<Object[]> rows = new ArrayList<>();
+ for (long index = 0; index < 4096; index++) {
+ rows.add(new Object[] {index / 64, 10000L + index});
+ }
+ return new
ListBatchCursor(List.of(RecordBatches.ofArrays(TOPOLOGY_SCHEMA, rows)));
+ }
+
+ private static List<Long> destinations(ReadResult result) throws
IOException {
+ List<Long> destinations = new ArrayList<>();
+ try (BatchCursor cursor = result.cursor()) {
+ while (cursor.next()) {
+ RecordBatch batch = cursor.batch();
+ for (int index = 0; index < batch.rowCount(); index++) {
+ destinations.add((Long) batch.column(0).getObject(index));
+ }
+ }
+ }
+ return destinations;
+ }
+
+ private static final class ListBatchCursor implements BatchCursor {
+ private final List<RecordBatch> batches;
+ private int index = -1;
+
+ private ListBatchCursor(List<RecordBatch> batches) {
+ this.batches = List.copyOf(batches);
+ }
+
+ @Override
+ public boolean next() {
+ index++;
+ return index < batches.size();
+ }
+
+ @Override
+ public RecordBatch batch() {
+ if (index < 0 || index >= batches.size()) {
+ throw new IllegalStateException("No current batch. Call next()
before batch().");
+ }
+ return batches.get(index);
+ }
+
+ @Override
+ public void close() {}
+ }
+}
diff --git
a/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetSpecConformanceTest.java
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetSpecConformanceTest.java
new file mode 100644
index 00000000..a36cf86a
--- /dev/null
+++
b/maven-projects/io-parquet/src/test/java/org/apache/graphar/io/parquet/ParquetSpecConformanceTest.java
@@ -0,0 +1,292 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.graphar.io.BatchCursor;
+import org.apache.graphar.io.ColumnType;
+import org.apache.graphar.io.Field;
+import org.apache.graphar.io.ReadRequest;
+import org.apache.graphar.io.ReadResult;
+import org.apache.graphar.io.RecordBatch;
+import org.apache.graphar.io.RecordBatches;
+import org.apache.graphar.io.Schema;
+import org.apache.graphar.io.WriteMode;
+import org.apache.graphar.io.WriteRequest;
+import org.apache.graphar.storage.InputFile;
+import org.apache.graphar.storage.OutputFile;
+import org.apache.graphar.storage.SeekableInput;
+import org.apache.graphar.storage.Storage;
+import org.apache.graphar.storage.local.LocalStorage;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.apache.parquet.schema.Type;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/** Verifies the Parquet layouts this module writes and accepts follow the
Parquet format spec. */
+public class ParquetSpecConformanceTest {
+ private Path directory;
+
+ @Before
+ public void createDirectory() throws IOException {
+ directory = Files.createTempDirectory("graphar-parquet-spec-");
+ }
+
+ @After
+ public void deleteDirectory() throws IOException {
+ try (var paths = Files.list(directory)) {
+ for (Path path : (Iterable<Path>) paths::iterator) {
+ Files.delete(path);
+ }
+ }
+ Files.delete(directory);
+ }
+
+ @Test
+ public void writesThreeLevelListsWithOptionalElementsAndNullEntries()
throws IOException {
+ Schema schema =
+ new Schema(
+ List.of(
+ new Field(
+ "tags",
+
ColumnType.listOf(ColumnType.of(ColumnType.Kind.STRING)),
+ true)));
+ URI uri = write(schema, List.of(List.of(Arrays.asList("a", null,
"b"))));
+
+ Type element = listElement(uri, "tags");
+ assertEquals("element", element.getName());
+ assertTrue(element.isRepetition(Type.Repetition.OPTIONAL));
+
+ ReadResult result = read(uri);
+ assertEquals(schema, result.cursor().next() ?
result.cursor().batch().schema() : null);
+ assertEquals(Arrays.asList("a", null, "b"),
result.cursor().batch().column(0).getObject(0));
+ result.cursor().close();
+ }
+
+ @Test
+ public void writesRequiredElementsForANonNullableElementList() throws
IOException {
+ Schema schema =
+ new Schema(
+ List.of(
+ new Field(
+ "ids",
+ ColumnType.listOfElement(
+ new Field(
+ "element",
+
ColumnType.of(ColumnType.Kind.INT64),
+ false)),
+ false)));
+ URI uri = write(schema, List.of(List.of(List.of(1L, 2L))));
+
+ assertTrue(listElement(uri,
"ids").isRepetition(Type.Repetition.REQUIRED));
+ assertEquals(List.of(List.of(1L, 2L)), column(uri, 0));
+ }
+
+ @Test
+ public void readsAListWrittenByAnotherParquetWriter() throws IOException {
+ MessageType foreign =
+ MessageTypeParser.parseMessageType(
+ "message spark_schema {"
+ + " optional group scores (LIST) {"
+ + " repeated group list {"
+ + " optional int64 element;"
+ + " }"
+ + " }"
+ + "}");
+ URI uri = directory.resolve("foreign.parquet").toUri();
+ LocalStorage storage = new LocalStorage();
+ try (ParquetWriter<Group> writer =
+ ExampleParquetWriter.builder(new
ParquetOutputFile(storage.outputFile(uri)))
+ .withType(foreign)
+ .build()) {
+ Group row = new SimpleGroupFactory(foreign).newGroup();
+ Group list = row.addGroup("scores");
+ list.addGroup("list").add("element", 7L);
+ list.addGroup("list");
+ writer.write(row);
+ }
+
+ assertEquals(List.of(Arrays.asList(7L, null)), column(uri, 0));
+ }
+
+ @Test
+ public void releasesTheFileAsSoonAsTheLimitIsReached() throws IOException {
+ Schema schema =
+ new Schema(List.of(new Field("id",
ColumnType.of(ColumnType.Kind.INT64), false)));
+ List<List<Object>> rows = new ArrayList<>();
+ for (long id = 0; id < 10; id++) {
+ rows.add(List.of(id));
+ }
+ URI uri = write(schema, rows);
+ TrackingStorage storage = new TrackingStorage(new LocalStorage());
+
+ BatchCursor cursor =
+ new ParquetPhysicalReader(storage)
+ .read(ReadRequest.builder(uri).limit(3).build())
+ .cursor();
+
+ assertTrue(cursor.next());
+ assertEquals(3, cursor.batch().rowCount());
+ assertEquals(storage.opened, storage.closed);
+ assertTrue(storage.opened > 0);
+ assertEquals(2L, cursor.batch().column(0).getObject(2));
+ assertFalse(cursor.next());
+ cursor.close();
+ }
+
+ private URI write(Schema schema, List<? extends List<?>> rows) throws
IOException {
+ URI uri = Files.createTempFile(directory, "spec-", ".parquet").toUri();
+ Files.delete(Path.of(uri));
+ RecordBatch batch = RecordBatches.ofRows(schema, rows);
+ new ParquetPhysicalWriter(new LocalStorage())
+ .write(new WriteRequest(uri, schema, WriteMode.CREATE_NEW),
new OneBatch(batch));
+ return uri;
+ }
+
+ private static ReadResult read(URI uri) throws IOException {
+ return new ParquetPhysicalReader(new
LocalStorage()).read(ReadRequest.builder(uri).build());
+ }
+
+ private static List<Object> column(URI uri, int column) throws IOException
{
+ List<Object> values = new ArrayList<>();
+ try (BatchCursor cursor = read(uri).cursor()) {
+ while (cursor.next()) {
+ RecordBatch batch = cursor.batch();
+ for (int row = 0; row < batch.rowCount(); row++) {
+ values.add(batch.column(column).getObject(row));
+ }
+ }
+ }
+ return values;
+ }
+
+ private static Type listElement(URI uri, String name) throws IOException {
+ InputFile file = new LocalStorage().inputFile(uri);
+ try (ParquetFileReader reader = ParquetFileReader.open(new
ParquetInputFile(file))) {
+ MessageType schema =
reader.getFooter().getFileMetaData().getSchema();
+ return
schema.getType(name).asGroupType().getType(0).asGroupType().getType(0);
+ }
+ }
+
+ private static final class OneBatch implements BatchCursor {
+ private final RecordBatch batch;
+ private boolean advanced;
+
+ private OneBatch(RecordBatch batch) {
+ this.batch = batch;
+ }
+
+ @Override
+ public boolean next() {
+ boolean first = !advanced;
+ advanced = true;
+ return first;
+ }
+
+ @Override
+ public RecordBatch batch() {
+ return batch;
+ }
+
+ @Override
+ public void close() {}
+ }
+
+ private static final class TrackingStorage implements Storage {
+ private final Storage delegate;
+ private int opened;
+ private int closed;
+
+ private TrackingStorage(Storage delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public InputFile inputFile(URI uri) {
+ InputFile file = delegate.inputFile(uri);
+ return new InputFile() {
+ @Override
+ public URI uri() {
+ return file.uri();
+ }
+
+ @Override
+ public long size() throws IOException {
+ return file.size();
+ }
+
+ @Override
+ public SeekableInput open() throws IOException {
+ SeekableInput input = file.open();
+ opened++;
+ return new SeekableInput() {
+ @Override
+ public long position() throws IOException {
+ return input.position();
+ }
+
+ @Override
+ public void seek(long newPosition) throws IOException {
+ input.seek(newPosition);
+ }
+
+ @Override
+ public int read(ByteBuffer destination) throws
IOException {
+ return input.read(destination);
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed++;
+ input.close();
+ }
+ };
+ }
+ };
+ }
+
+ @Override
+ public OutputFile outputFile(URI uri) {
+ return delegate.outputFile(uri);
+ }
+
+ @Override
+ public boolean exists(URI uri) throws IOException {
+ return delegate.exists(uri);
+ }
+ }
+}
diff --git a/maven-projects/pom.xml b/maven-projects/pom.xml
index 1ccf661b..704e0512 100644
--- a/maven-projects/pom.xml
+++ b/maven-projects/pom.xml
@@ -81,6 +81,7 @@
<module>storage-api</module>
<module>storage-local</module>
<module>io-api</module>
+ <module>io-parquet</module>
<module>storage-s3</module>
<module>core</module>
</modules>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
