Implementations are selected once per column based on the column's type, + * nullability, and encoding configuration. Each implementation represents + * the most efficient write path for that combination. + */ +public interface ArrowColumnWriter { + + /** + * Writes values from the given Arrow vector (rows {@code offset} to {@code offset + length - 1}) + * to the underlying Parquet page writer. + * + * @param vector the Arrow vector containing column values + * @param offset the first row index to write (inclusive) + * @param length the number of rows to write + * @throws IOException if an I/O error occurs during page writing + */ + void write(FieldVector vector, int offset, int length) throws IOException; +} diff --git a/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/ArrowColumnWriterFactory.java b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/ArrowColumnWriterFactory.java new file mode 100644 index 0000000000..56cb74fa67 --- /dev/null +++ b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/ArrowColumnWriterFactory.java @@ -0,0 +1,107 @@ +/* + * 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.parquet.arrow.writer; + +import org.apache.parquet.column.page.PageWriter; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Type; + +/** + * Factory that selects the optimal {@link ArrowColumnWriter} strategy for each column + * based on the Parquet schema (type, nullability, encoding). + * + *
Strategy selection (best to worst): + *
Unlike the standard {@code ParquetWriter Usage:
+ * This writer is NOT thread-safe.
+ */
+public class ArrowParquetWriter implements Closeable {
+
+ private static final long DEFAULT_ROW_GROUP_SIZE = 128 * 1024 * 1024L;
+
+ /** Target page size. Pages larger than this reduce column-index predicate pushdown granularity. */
+ private static final int TARGET_PAGE_SIZE_BYTES = 1024 * 1024;
+
+ /** Default byte estimate for variable-width columns when computing rows-per-page. */
+ private static final int VAR_WIDTH_ESTIMATE_BYTES = 32;
+
+ private final ParquetFileWriter fileWriter;
+ private final MessageType schema;
+ private final ParquetProperties props;
+ private final Function Both Arrow and Parquet store booleans as bit-packed values (1 bit per value).
+ * Arrow uses LSB bit ordering within each byte, which matches Parquet's boolean
+ * PLAIN encoding. For non-null columns, the data buffer can be used directly.
+ * For nullable columns, non-null values are compacted.
+ */
+class BooleanPlainWriter implements ArrowColumnWriter {
+
+ private final PageWriter pageWriter;
+ private final PrimitiveType type;
+ private final int maxDefinitionLevel;
+ private final boolean isNullable;
+
+ BooleanPlainWriter(PageWriter pageWriter, PrimitiveType type, int maxDefinitionLevel,
+ boolean isNullable) {
+ this.pageWriter = pageWriter;
+ this.type = type;
+ this.maxDefinitionLevel = maxDefinitionLevel;
+ this.isNullable = isNullable;
+ }
+
+ @Override
+ public void write(FieldVector vector, int offset, int length) throws IOException {
+ BitVector bitVector = (BitVector) vector;
+ ArrowBuf validityBuf = vector.getValidityBuffer();
+ ArrowBuf dataBuf = bitVector.getDataBuffer();
+
+ int nullCount = 0;
+ int trueCount = 0;
+
+ if (isNullable) {
+ for (int i = 0; i < length; i++) {
+ if (isNull(validityBuf, offset + i)) {
+ nullCount++;
+ } else if (getBit(dataBuf, offset + i)) {
+ trueCount++;
+ }
+ }
+ } else {
+ for (int i = 0; i < length; i++) {
+ if (getBit(dataBuf, offset + i)) {
+ trueCount++;
+ }
+ }
+ }
+
+ int nonNullCount = length - nullCount;
+
+ // Produce bit-packed boolean data for non-null values
+ int byteCount = (nonNullCount + 7) / 8;
+ byte[] booleanBytes = new byte[byteCount];
+ int bitPos = 0;
+
+ for (int i = 0; i < length; i++) {
+ if (isNullable && isNull(validityBuf, offset + i)) {
+ continue;
+ }
+ if (getBit(dataBuf, offset + i)) {
+ booleanBytes[bitPos / 8] |= (1 << (bitPos % 8));
+ }
+ bitPos++;
+ }
+
+ // Statistics
+ BooleanStatistics stats = (BooleanStatistics) Statistics.createStats(type);
+ stats.setNumNulls(nullCount);
+ if (nonNullCount > 0) {
+ boolean hasTrue = trueCount > 0;
+ boolean hasFalse = (nonNullCount - trueCount) > 0;
+ if (hasTrue && hasFalse) {
+ stats.updateStats(true);
+ stats.updateStats(false);
+ } else if (hasTrue) {
+ stats.updateStats(true);
+ } else {
+ stats.updateStats(false);
+ }
+ }
+
+ // Repetition levels
+ BytesInput rl = LevelEncoder.encodeConstant(0, length, 0);
+
+ // Definition levels
+ BytesInput dl;
+ if (isNullable) {
+ dl = LevelEncoder.encodeFromValidityBitmap(validityBuf, offset, length, maxDefinitionLevel);
+ } else {
+ dl = LevelEncoder.encodeConstant(maxDefinitionLevel, length, maxDefinitionLevel);
+ }
+
+ pageWriter.writePage(
+ BytesInput.concat(rl, dl, BytesInput.from(booleanBytes)),
+ length,
+ length,
+ stats,
+ Encoding.RLE,
+ Encoding.RLE,
+ Encoding.PLAIN);
+ }
+
+ private static boolean getBit(ArrowBuf buf, int index) {
+ int byteIndex = index >> 3;
+ int bitIndex = index & 7;
+ return ((buf.getByte(byteIndex) >> bitIndex) & 1) == 1;
+ }
+
+ private static boolean isNull(ArrowBuf validityBuf, int index) {
+ if (validityBuf == null || validityBuf.capacity() == 0) {
+ return false;
+ }
+ int byteIndex = index >> 3;
+ int bitIndex = index & 7;
+ return ((validityBuf.getByte(byteIndex) >> bitIndex) & 1) == 0;
+ }
+}
diff --git a/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/LevelEncoder.java b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/LevelEncoder.java
new file mode 100644
index 0000000000..7a652ae5dd
--- /dev/null
+++ b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/LevelEncoder.java
@@ -0,0 +1,115 @@
+/*
+ * 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.parquet.arrow.writer;
+
+import java.io.IOException;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridEncoder;
+
+/**
+ * Encodes Parquet repetition and definition levels for flat schemas.
+ *
+ * For flat schemas, repetition levels are always 0 and definition levels are either
+ * 0 (null) or max (present). These produce single-value RLE runs encoded in O(1).
+ */
+final class LevelEncoder {
+
+ private LevelEncoder() {}
+
+ /**
+ * Encodes a constant level value repeated {@code count} times in O(1).
+ *
+ * Produces a single RLE run. The RLE/Bit-Packing Hybrid format for a single run is:
+ * V1 data pages require a 4-byte little-endian length prefix before the level data.
+ *
+ * @param value the constant level value to repeat
+ * @param count the number of values
+ * @param maxLevel the maximum possible level (determines bit width)
+ * @return encoded level bytes with V1 length prefix, or empty if maxLevel is 0
+ */
+ static BytesInput encodeConstant(int value, int count, int maxLevel) {
+ if (maxLevel == 0) {
+ // Bit width 0: no level data written (reader knows it's always 0)
+ return BytesInput.empty();
+ }
+
+ int bitWidth = getBitWidth(maxLevel);
+ int valueByteCount = (bitWidth + 7) / 8;
+
+ // RLE header varint: (count << 1) encodes the run length with mode bit = 0 (RLE)
+ BytesInput headerBytes = BytesInput.fromUnsignedVarInt(count << 1);
+
+ // RLE value: little-endian encoding of the repeated value
+ byte[] valueEncoded = new byte[valueByteCount];
+ for (int i = 0; i < valueByteCount; i++) {
+ valueEncoded[i] = (byte) ((value >>> (i * 8)) & 0xFF);
+ }
+ BytesInput valueBytes = BytesInput.from(valueEncoded);
+
+ // The RLE payload (header + value)
+ BytesInput rlePayload = BytesInput.concat(headerBytes, valueBytes);
+
+ // V1 format: [4-byte LE length of level data][level data]
+ return BytesInput.concat(BytesInput.fromInt((int) rlePayload.size()), rlePayload);
+ }
+
+ /**
+ * Encodes definition levels from an Arrow validity bitmap.
+ *
+ * Uses the standard RLE encoder which naturally produces efficient runs for
+ * data with clustered nulls/non-nulls.
+ *
+ * V1 data pages require a 4-byte little-endian length prefix before the level data.
+ *
+ * @param validityBuf the Arrow validity buffer (bit i = 1 means value present)
+ * @param offset the starting row index in the validity buffer
+ * @param length the number of rows to encode
+ * @param maxDefinitionLevel the DL value for non-null rows
+ * @return encoded level bytes with V1 length prefix
+ * @throws IOException if encoding fails
+ */
+ static BytesInput encodeFromValidityBitmap(
+ org.apache.arrow.memory.ArrowBuf validityBuf, int offset, int length,
+ int maxDefinitionLevel) throws IOException {
+ int bitWidth = getBitWidth(maxDefinitionLevel);
+ RunLengthBitPackingHybridEncoder encoder = new RunLengthBitPackingHybridEncoder(
+ bitWidth, length, length,
+ org.apache.parquet.bytes.HeapByteBufferAllocator.getInstance());
+
+ for (int i = 0; i < length; i++) {
+ int byteIndex = (offset + i) >> 3;
+ int bitIndex = (offset + i) & 7;
+ boolean isSet = ((validityBuf.getByte(byteIndex) >> bitIndex) & 1) == 1;
+ encoder.writeInt(isSet ? maxDefinitionLevel : 0);
+ }
+
+ BytesInput encoded = encoder.toBytes();
+ return BytesInput.concat(BytesInput.fromInt((int) encoded.size()), encoded);
+ }
+
+ /** Returns the minimum number of bits needed to represent values up to maxLevel. */
+ private static int getBitWidth(int maxLevel) {
+ return 32 - Integer.numberOfLeadingZeros(maxLevel);
+ }
+}
diff --git a/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/NullablePlainWriter.java b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/NullablePlainWriter.java
new file mode 100644
index 0000000000..4335d34268
--- /dev/null
+++ b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/NullablePlainWriter.java
@@ -0,0 +1,135 @@
+/*
+ * 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.parquet.arrow.writer;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.PageWriter;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.schema.PrimitiveType;
+
+/**
+ * Column writer for nullable, fixed-width, PLAIN-encoded columns.
+ *
+ * Single-pass: iterates values once, copying non-null values to a compacted buffer,
+ * computing statistics, and counting nulls simultaneously.
+ *
+ * Conditions: PLAIN encoding, column is optional (nullable), type is fixed-width.
+ */
+class NullablePlainWriter implements ArrowColumnWriter {
+
+ private final PageWriter pageWriter;
+ private final PrimitiveType type;
+ private final int typeWidthBytes;
+ private final int maxDefinitionLevel;
+
+ NullablePlainWriter(PageWriter pageWriter, PrimitiveType type, int typeWidthBytes,
+ int maxDefinitionLevel) {
+ this.pageWriter = pageWriter;
+ this.type = type;
+ this.typeWidthBytes = typeWidthBytes;
+ this.maxDefinitionLevel = maxDefinitionLevel;
+ }
+
+ @Override
+ public void write(FieldVector vector, int offset, int length) throws IOException {
+ ArrowBuf dataBuf = vector.getDataBuffer();
+ ArrowBuf validityBuf = vector.getValidityBuffer();
+
+ // Allocate output at max possible size (all non-null). Single pass fills it.
+ ByteBuffer compactedData = ByteBuffer.allocate(length * typeWidthBytes);
+ compactedData.order(ByteOrder.LITTLE_ENDIAN);
+
+ ByteBuffer srcView = dataBuf.nioBuffer(
+ (long) offset * typeWidthBytes, (int) ((long) length * typeWidthBytes));
+ srcView.order(ByteOrder.LITTLE_ENDIAN);
+
+ // Single-pass: compact non-null values + compute stats + count nulls
+ Statistics stats = Statistics.createStats(type);
+ int nullCount = 0;
+
+ for (int i = 0; i < length; i++) {
+ if (isNull(validityBuf, offset + i)) {
+ nullCount++;
+ } else {
+ int srcPos = i * typeWidthBytes;
+ // Copy value bytes
+ for (int b = 0; b < typeWidthBytes; b++) {
+ compactedData.put(srcView.get(srcPos + b));
+ }
+ // Update stats from the source buffer
+ updateStats(stats, srcView, srcPos);
+ }
+ }
+
+ stats.setNumNulls(nullCount);
+ compactedData.flip();
+
+ // Repetition levels: all 0 (flat schema)
+ BytesInput rl = LevelEncoder.encodeConstant(0, length, 0);
+
+ // Definition levels: from validity bitmap
+ BytesInput dl = LevelEncoder.encodeFromValidityBitmap(
+ validityBuf, offset, length, maxDefinitionLevel);
+
+ // Write page
+ pageWriter.writePage(
+ BytesInput.concat(rl, dl, BytesInput.from(compactedData)),
+ length,
+ length,
+ stats,
+ Encoding.RLE,
+ Encoding.RLE,
+ Encoding.PLAIN);
+ }
+
+ private void updateStats(Statistics stats, ByteBuffer buf, int pos) {
+ switch (type.getPrimitiveTypeName()) {
+ case INT32:
+ stats.updateStats(buf.getInt(pos));
+ break;
+ case INT64:
+ stats.updateStats(buf.getLong(pos));
+ break;
+ case FLOAT:
+ stats.updateStats(buf.getFloat(pos));
+ break;
+ case DOUBLE:
+ stats.updateStats(buf.getDouble(pos));
+ break;
+ default:
+ // For other fixed-width types, skip stats (still produces valid file)
+ break;
+ }
+ }
+
+ private static boolean isNull(ArrowBuf validityBuf, int index) {
+ if (validityBuf == null || validityBuf.capacity() == 0) {
+ return false; // No validity buffer means all values are non-null
+ }
+ int byteIndex = index >> 3;
+ int bitIndex = index & 7;
+ return ((validityBuf.getByte(byteIndex) >> bitIndex) & 1) == 0;
+ }
+}
diff --git a/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/StatsComputer.java b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/StatsComputer.java
new file mode 100644
index 0000000000..2b5356ae90
--- /dev/null
+++ b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/StatsComputer.java
@@ -0,0 +1,169 @@
+/*
+ * 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.parquet.arrow.writer;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.parquet.column.statistics.FloatStatistics;
+import org.apache.parquet.column.statistics.DoubleStatistics;
+import org.apache.parquet.column.statistics.IntStatistics;
+import org.apache.parquet.column.statistics.LongStatistics;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.schema.PrimitiveType;
+
+/**
+ * Computes Parquet page statistics from Arrow data buffers in a single sequential scan.
+ * Thread-safe: all state is returned via {@link StatsResult}, no mutable shared fields.
+ */
+final class StatsComputer {
+
+ private StatsComputer() {}
+
+ /** Holds computed statistics plus NaN count for floating-point columns. */
+ static final class StatsResult {
+ final Statistics statistics;
+ final int nanCount;
+
+ StatsResult(Statistics statistics, int nanCount) {
+ this.statistics = statistics;
+ this.nanCount = nanCount;
+ }
+ }
+
+ /**
+ * Computes statistics for a non-null fixed-width column.
+ *
+ * @param dataBuf the Arrow data buffer
+ * @param offset the starting row index
+ * @param length the number of values
+ * @param type the Parquet primitive type
+ * @return statistics and NaN count (0 for non-floating-point types)
+ */
+ static StatsResult compute(ArrowBuf dataBuf, int offset, int length, PrimitiveType type) {
+ switch (type.getPrimitiveTypeName()) {
+ case INT32:
+ return computeIntStats(dataBuf, offset, length, type);
+ case INT64:
+ return computeLongStats(dataBuf, offset, length, type);
+ case FLOAT:
+ return computeFloatStats(dataBuf, offset, length, type);
+ case DOUBLE:
+ return computeDoubleStats(dataBuf, offset, length, type);
+ default:
+ Statistics stats = Statistics.createStats(type);
+ stats.setNumNulls(0);
+ return new StatsResult(stats, 0);
+ }
+ }
+
+ private static StatsResult computeIntStats(
+ ArrowBuf dataBuf, int offset, int length, PrimitiveType type) {
+ ByteBuffer buf = dataBuf.nioBuffer(
+ (long) offset * Integer.BYTES, (int) ((long) length * Integer.BYTES));
+ buf.order(ByteOrder.LITTLE_ENDIAN);
+
+ int min = Integer.MAX_VALUE;
+ int max = Integer.MIN_VALUE;
+ for (int i = 0; i < length; i++) {
+ int v = buf.getInt();
+ if (v < min) min = v;
+ if (v > max) max = v;
+ }
+
+ IntStatistics stats = (IntStatistics) Statistics.createStats(type);
+ stats.setMinMax(min, max);
+ stats.setNumNulls(0);
+ return new StatsResult(stats, 0);
+ }
+
+ private static StatsResult computeLongStats(
+ ArrowBuf dataBuf, int offset, int length, PrimitiveType type) {
+ ByteBuffer buf = dataBuf.nioBuffer(
+ (long) offset * Long.BYTES, (int) ((long) length * Long.BYTES));
+ buf.order(ByteOrder.LITTLE_ENDIAN);
+
+ long min = Long.MAX_VALUE;
+ long max = Long.MIN_VALUE;
+ for (int i = 0; i < length; i++) {
+ long v = buf.getLong();
+ if (v < min) min = v;
+ if (v > max) max = v;
+ }
+
+ LongStatistics stats = (LongStatistics) Statistics.createStats(type);
+ stats.setMinMax(min, max);
+ stats.setNumNulls(0);
+ return new StatsResult(stats, 0);
+ }
+
+ private static StatsResult computeFloatStats(
+ ArrowBuf dataBuf, int offset, int length, PrimitiveType type) {
+ ByteBuffer buf = dataBuf.nioBuffer(
+ (long) offset * Float.BYTES, (int) ((long) length * Float.BYTES));
+ buf.order(ByteOrder.LITTLE_ENDIAN);
+
+ float min = Float.POSITIVE_INFINITY;
+ float max = Float.NEGATIVE_INFINITY;
+ int nanCount = 0;
+ for (int i = 0; i < length; i++) {
+ float v = buf.getFloat();
+ if (Float.isNaN(v)) {
+ nanCount++;
+ } else {
+ if (Float.compare(v, min) < 0) min = v;
+ if (Float.compare(v, max) > 0) max = v;
+ }
+ }
+
+ FloatStatistics stats = (FloatStatistics) Statistics.createStats(type);
+ if (length - nanCount > 0) {
+ stats.setMinMax(min, max);
+ }
+ stats.setNumNulls(0);
+ return new StatsResult(stats, nanCount);
+ }
+
+ private static StatsResult computeDoubleStats(
+ ArrowBuf dataBuf, int offset, int length, PrimitiveType type) {
+ ByteBuffer buf = dataBuf.nioBuffer(
+ (long) offset * Double.BYTES, (int) ((long) length * Double.BYTES));
+ buf.order(ByteOrder.LITTLE_ENDIAN);
+
+ double min = Double.POSITIVE_INFINITY;
+ double max = Double.NEGATIVE_INFINITY;
+ int nanCount = 0;
+ for (int i = 0; i < length; i++) {
+ double v = buf.getDouble();
+ if (Double.isNaN(v)) {
+ nanCount++;
+ } else {
+ if (Double.compare(v, min) < 0) min = v;
+ if (Double.compare(v, max) > 0) max = v;
+ }
+ }
+
+ DoubleStatistics stats = (DoubleStatistics) Statistics.createStats(type);
+ if (length - nanCount > 0) {
+ stats.setMinMax(min, max);
+ }
+ stats.setNumNulls(0);
+ return new StatsResult(stats, nanCount);
+ }
+}
diff --git a/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/VarWidthPlainWriter.java b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/VarWidthPlainWriter.java
new file mode 100644
index 0000000000..f1947b57d7
--- /dev/null
+++ b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/VarWidthPlainWriter.java
@@ -0,0 +1,139 @@
+/*
+ * 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.parquet.arrow.writer;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.PageWriter;
+import org.apache.parquet.column.statistics.BinaryStatistics;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.PrimitiveType;
+
+/**
+ * Column writer for variable-width types (STRING, BINARY) using PLAIN encoding.
+ *
+ * Arrow stores variable-width data as an offset buffer (int32[N+1]) plus a contiguous data
+ * buffer. Parquet PLAIN encoding stores each value as [4-byte length][value bytes]. This writer
+ * transforms between the two layouts in a single pass.
+ *
+ * Handles both nullable and non-null columns via the Arrow validity bitmap.
+ */
+class VarWidthPlainWriter implements ArrowColumnWriter {
+
+ private final PageWriter pageWriter;
+ private final PrimitiveType type;
+ private final int maxDefinitionLevel;
+ private final boolean isNullable;
+
+ VarWidthPlainWriter(PageWriter pageWriter, PrimitiveType type, int maxDefinitionLevel,
+ boolean isNullable) {
+ this.pageWriter = pageWriter;
+ this.type = type;
+ this.maxDefinitionLevel = maxDefinitionLevel;
+ this.isNullable = isNullable;
+ }
+
+ @Override
+ public void write(FieldVector vector, int offset, int length) throws IOException {
+ BaseVariableWidthVector varVector = (BaseVariableWidthVector) vector;
+ ArrowBuf offsetBuf = varVector.getOffsetBuffer();
+ ArrowBuf dataBuf = varVector.getDataBuffer();
+ ArrowBuf validityBuf = vector.getValidityBuffer();
+
+ // Calculate total data size for non-null values
+ int nullCount = 0;
+ int totalDataBytes = 0;
+ for (int i = 0; i < length; i++) {
+ if (isNullable && isNull(validityBuf, offset + i)) {
+ nullCount++;
+ } else {
+ int start = offsetBuf.getInt((long) (offset + i) * Integer.BYTES);
+ int end = offsetBuf.getInt((long) (offset + i + 1) * Integer.BYTES);
+ totalDataBytes += (end - start);
+ }
+ }
+
+ int nonNullCount = length - nullCount;
+ // Parquet PLAIN for binary: [4-byte length][data] per value
+ int pageDataSize = nonNullCount * Integer.BYTES + totalDataBytes;
+ ByteBuffer pageData = ByteBuffer.allocate(pageDataSize);
+ pageData.order(ByteOrder.LITTLE_ENDIAN);
+
+ // Build statistics while writing
+ BinaryStatistics stats = (BinaryStatistics) Statistics.createStats(type);
+ stats.setNumNulls(nullCount);
+
+ for (int i = 0; i < length; i++) {
+ if (isNullable && isNull(validityBuf, offset + i)) {
+ continue; // skip null values in data section
+ }
+ int start = offsetBuf.getInt((long) (offset + i) * Integer.BYTES);
+ int end = offsetBuf.getInt((long) (offset + i + 1) * Integer.BYTES);
+ int len = end - start;
+
+ // Write length-prefixed value
+ pageData.putInt(len);
+ if (len > 0) {
+ byte[] valueBytes = new byte[len];
+ dataBuf.getBytes(start, valueBytes);
+ pageData.put(valueBytes);
+ stats.updateStats(Binary.fromReusedByteArray(valueBytes));
+ } else {
+ stats.updateStats(Binary.EMPTY);
+ }
+ }
+ pageData.flip();
+
+ // Repetition levels: all 0 (flat schema)
+ BytesInput rl = LevelEncoder.encodeConstant(0, length, 0);
+
+ // Definition levels
+ BytesInput dl;
+ if (isNullable) {
+ dl = LevelEncoder.encodeFromValidityBitmap(validityBuf, offset, length, maxDefinitionLevel);
+ } else {
+ dl = LevelEncoder.encodeConstant(maxDefinitionLevel, length, maxDefinitionLevel);
+ }
+
+ pageWriter.writePage(
+ BytesInput.concat(rl, dl, BytesInput.from(pageData)),
+ length,
+ length,
+ stats,
+ Encoding.RLE,
+ Encoding.RLE,
+ Encoding.PLAIN);
+ }
+
+ private static boolean isNull(ArrowBuf validityBuf, int index) {
+ if (validityBuf == null || validityBuf.capacity() == 0) {
+ return false;
+ }
+ int byteIndex = index >> 3;
+ int bitIndex = index & 7;
+ return ((validityBuf.getByte(byteIndex) >> bitIndex) & 1) == 0;
+ }
+}
diff --git a/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/ZeroCopyPlainWriter.java b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/ZeroCopyPlainWriter.java
new file mode 100644
index 0000000000..ff56105853
--- /dev/null
+++ b/parquet-arrow/src/main/java/org/apache/parquet/arrow/writer/ZeroCopyPlainWriter.java
@@ -0,0 +1,89 @@
+/*
+ * 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.parquet.arrow.writer;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.PageWriter;
+import org.apache.parquet.schema.PrimitiveType;
+
+/**
+ * Zero-copy column writer for non-null, fixed-width, PLAIN-encoded columns.
+ *
+ * Arrow's data buffer for fixed-width types (Int32, Int64, Float, Double) is stored as
+ * contiguous little-endian values — which is identical to Parquet's PLAIN encoding. This writer
+ * wraps the Arrow buffer directly as a Parquet page without copying or transforming any bytes.
+ *
+ * Conditions: PLAIN encoding, column is required (non-null), type is fixed-width.
+ */
+class ZeroCopyPlainWriter implements ArrowColumnWriter {
+
+ private final PageWriter pageWriter;
+ private final PrimitiveType type;
+ private final int typeWidthBytes;
+ private final int maxDefinitionLevel;
+
+ /**
+ * @param pageWriter the Parquet page writer for this column
+ * @param type the Parquet primitive type descriptor
+ * @param typeWidthBytes the byte width of each value (4 for INT32, 8 for INT64/FLOAT64, etc.)
+ * @param maxDefinitionLevel the max definition level for this column
+ */
+ ZeroCopyPlainWriter(PageWriter pageWriter, PrimitiveType type, int typeWidthBytes,
+ int maxDefinitionLevel) {
+ this.pageWriter = pageWriter;
+ this.type = type;
+ this.typeWidthBytes = typeWidthBytes;
+ this.maxDefinitionLevel = maxDefinitionLevel;
+ }
+
+ @Override
+ public void write(FieldVector vector, int offset, int length) throws IOException {
+ ArrowBuf dataBuf = vector.getDataBuffer();
+
+ // Data: wrap Arrow's data buffer directly — zero copy
+ long startByte = (long) offset * typeWidthBytes;
+ long lengthBytes = (long) length * typeWidthBytes;
+ ByteBuffer nioBuffer = dataBuf.nioBuffer(startByte, (int) lengthBytes);
+ BytesInput data = BytesInput.from(nioBuffer);
+
+ // Repetition levels: all 0 for flat schema — encode as single RLE run
+ BytesInput rl = LevelEncoder.encodeConstant(0, length, 0);
+
+ // Definition levels: all max for non-null — encode as single RLE run
+ BytesInput dl = LevelEncoder.encodeConstant(maxDefinitionLevel, length, maxDefinitionLevel);
+
+ // Statistics: sequential scan of the buffer
+ StatsComputer.StatsResult statsResult = StatsComputer.compute(dataBuf, offset, length, type);
+
+ // Write the assembled page
+ pageWriter.writePage(
+ BytesInput.concat(rl, dl, data),
+ length, // valueCount
+ length, // rowCount (same for flat schema)
+ statsResult.statistics,
+ Encoding.RLE, // RL encoding
+ Encoding.RLE, // DL encoding
+ Encoding.PLAIN); // values encoding
+ }
+}
diff --git a/parquet-arrow/src/test/java/org/apache/parquet/arrow/writer/TestArrowParquetWriter.java b/parquet-arrow/src/test/java/org/apache/parquet/arrow/writer/TestArrowParquetWriter.java
new file mode 100644
index 0000000000..5fced4a9d1
--- /dev/null
+++ b/parquet-arrow/src/test/java/org/apache/parquet/arrow/writer/TestArrowParquetWriter.java
@@ -0,0 +1,539 @@
+/*
+ * 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.parquet.arrow.writer;
+
+import static java.util.Arrays.asList;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.Float4Vector;
+import org.apache.arrow.vector.Float8Vector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.FloatingPointPrecision;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.parquet.arrow.schema.SchemaConverter;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.example.GroupReadSupport;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.io.LocalInputFile;
+import org.apache.parquet.io.LocalOutputFile;
+import org.apache.parquet.io.OutputFile;
+import org.apache.parquet.schema.MessageType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class TestArrowParquetWriter {
+
+ @TempDir
+ File tempDir;
+
+ private BufferAllocator allocator;
+
+ @BeforeEach
+ void setUp() {
+ allocator = new RootAllocator();
+ }
+
+ @AfterEach
+ void tearDown() {
+ allocator.close();
+ }
+
+ @Test
+ void testWriteRequiredIntegers() throws IOException {
+ Schema arrowSchema = new Schema(asList(
+ new Field("id", FieldType.notNullable(new ArrowType.Int(32, true)), null),
+ new Field("value", FieldType.notNullable(new ArrowType.Int(64, true)), null)));
+
+ MessageType parquetSchema = new SchemaConverter().fromArrow(arrowSchema).getParquetSchema();
+ Path path = tempDir.toPath().resolve("required_ints.parquet");
+ OutputFile outputFile = new LocalOutputFile(path);
+
+ try (VectorSchemaRoot batch = VectorSchemaRoot.create(arrowSchema, allocator)) {
+ IntVector idVector = (IntVector) batch.getVector("id");
+ BigIntVector valueVector = (BigIntVector) batch.getVector("value");
+ idVector.allocateNew(3);
+ valueVector.allocateNew(3);
+ idVector.set(0, 1);
+ idVector.set(1, 2);
+ idVector.set(2, 3);
+ valueVector.set(0, 100L);
+ valueVector.set(1, 200L);
+ valueVector.set(2, 300L);
+ batch.setRowCount(3);
+
+ try (ArrowParquetWriter writer = new ArrowParquetWriter(outputFile, parquetSchema)) {
+ writer.writeBatch(batch);
+ }
+ }
+
+ try (ParquetReader{@code
+ * MessageType schema = new SchemaConverter().fromArrow(arrowSchema).getParquetSchema();
+ * try (ArrowParquetWriter writer = new ArrowParquetWriter(outputFile, schema)) {
+ * writer.writeBatch(batch);
+ * }
+ * }
+ *
+ *
+ *
+ *
+ *