@@ -29,6 +29,7 @@
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesUtils;
 import org.apache.parquet.column.values.ValuesReader;
 import org.apache.parquet.io.ParquetDecodingException;
 
@@ -54,8 +55,12 @@ abstract class AlpValuesReader extends ValuesReader {
   protected int pageValueIndex;
   protected int currentVectorIndex;
 
+  // One entry per vector plus a trailing entry for the end of the vector data, so the entry after
+  // any vector's offset is always where that vector ends.
   protected int[] vectorOffsets;
   protected ByteBuffer vectorsData;
+  // Captured once when the page is sliced, so nothing later depends on the buffer's position/limit.
+  protected int vectorDataSize;
   protected int offsetArraySize;
 
   // Scratch buffer for exception positions within a vector; shared by both readers (int[] in each).
@@ -86,7 +91,11 @@ public void initFromPage(int valuesCount, ByteBufferInputStream stream)
     int minVectorBytes = ALP_INFO_SIZE + forInfoSize();
     readOffsetArray(stream, minVectorBytes);
     sliceVectorData(stream);
-    validateVectorOffsets(minVectorBytes);
+    // The trailing entry is filled here rather than in readOffsetArray because the vector data
+    // size is only known once the data has been sliced.
+    vectorOffsets[numVectors] = offsetArraySize + vectorDataSize;
+    validateVectorOffsets(
+        vectorOffsets, minVectorBytes, maxVectorBytes(valuesPerVector, minVectorBytes, exceptionValueSize()));
 
     allocateDecodedBuffer(valuesPerVector);
     this.excPositionsBuffer = new int[valuesPerVector];
@@ -134,7 +143,10 @@ private void readPageHeader(ByteBufferInputStream stream, int valuesCount) throw
     this.currentVectorIndex = -1;
   }
 
-  /** Reads the offset array, one entry per vector, after bounding the vector count. */
+  /**
+   * Reads the offset array, one entry per vector, after bounding the vector count and checking the
+   * first offset. Leaves room for the trailing end-of-data entry, which the caller fills in.
+   */
   private void readOffsetArray(ByteBufferInputStream stream, int minVectorBytes) throws IOException {
     // Bound the vector count against the bytes actually present before allocating anything.
     // numElements is only capped by valuesCount, which a forged page can set as high as
@@ -154,8 +166,20 @@ private void readOffsetArray(ByteBufferInputStream stream, int minVectorBytes) t
 
     this.offsetArraySize = numVectors * Integer.BYTES;
     ByteBuffer offsetBuf = stream.slice(offsetArraySize).order(ByteOrder.LITTLE_ENDIAN);
-    this.vectorOffsets = new int[numVectors];
-    for (int v = 0; v < numVectors; v++) {
+    if (numVectors == 0) {
+      this.vectorOffsets = new int[1];
+      return;
+    }
+    // Offsets are measured from the start of the offset array, so the first vector must begin
+    // exactly where the array ends. Check that before allocating and decoding the rest.
+    int firstOffset = offsetBuf.getInt();
+    if (firstOffset != offsetArraySize) {
+      throw new ParquetDecodingException(
+          "ALP first vector offset " + firstOffset + " must equal the offset array size " + offsetArraySize);
+    }
+    this.vectorOffsets = new int[numVectors + 1];
+    vectorOffsets[0] = firstOffset;
+    for (int v = 1; v < numVectors; v++) {
       vectorOffsets[v] = offsetBuf.getInt();
     }
   }
@@ -165,48 +189,56 @@ private void readOffsetArray(ByteBufferInputStream stream, int minVectorBytes) t
    * absolute {@code get(pos)} reads.
    */
   private void sliceVectorData(ByteBufferInputStream stream) throws IOException {
-    int remainingBytes = (int) stream.available();
-    ByteBuffer rawSlice = stream.slice(remainingBytes);
+    this.vectorDataSize = (int) stream.available();
+    ByteBuffer rawSlice = stream.slice(vectorDataSize);
     this.vectorsData = rawSlice.slice().order(ByteOrder.LITTLE_ENDIAN);
   }
 
   /**
-   * Validates the offset array against the page body before any vector is decoded. Offsets are
-   * relative to the start of the compression body (the offset array itself), so the first vector
-   * must begin exactly where the offset array ends, offsets must increase by at least one vector's
-   * fixed headers, and every vector must start early enough to fit those headers. Without this a
-   * forged offset can point into another vector and silently decode the wrong values.
+   * Validates the offset array before any vector is decoded. {@code vectorOffsets} holds one entry
+   * per vector followed by the end of the vector data, all measured from the start of the offset
+   * array. The first entry is checked when the array is read, so this starts at entry 1: each
+   * vector, including the last, must span at least its fixed headers and no more than
+   * {@code maxVectorBytes}. Without this a forged offset can point into another vector and
+   * silently decode the wrong values.
    */
-  private void validateVectorOffsets(int minVectorBytes) {
-    int bodyLimit = offsetArraySize + vectorsData.limit();
-    for (int v = 0; v < numVectors; v++) {
-      int offset = vectorOffsets[v];
-      if (v == 0) {
-        if (offset != offsetArraySize) {
-          throw new ParquetDecodingException("ALP first vector offset " + offset
-              + " must equal the offset array size " + offsetArraySize);
-        }
-      } else if ((long) offset - vectorOffsets[v - 1] < minVectorBytes) {
-        throw new ParquetDecodingException("ALP vector offsets must increase by at least "
-            + minVectorBytes + " bytes, but vector " + v + " starts at " + offset + " after vector "
-            + (v - 1) + " at " + vectorOffsets[v - 1]);
+  static void validateVectorOffsets(int[] vectorOffsets, int minVectorBytes, int maxVectorBytes) {
+    int numVectors = vectorOffsets.length - 1;
+    for (int v = 1; v <= numVectors; v++) {
+      long span = (long) vectorOffsets[v] - vectorOffsets[v - 1];
+      if (span < minVectorBytes) {
+        throw new ParquetDecodingException("ALP vector " + (v - 1) + " at offset " + vectorOffsets[v - 1]
+            + " spans " + span + " bytes, fewer than the " + minVectorBytes + " its headers need");
       }
-      if (offset > bodyLimit - minVectorBytes) {
-        throw new ParquetDecodingException("ALP vector " + v + " offset " + offset
-            + " leaves no room for its headers before the end of the page body at " + bodyLimit);
+      if (span > maxVectorBytes) {
+        throw new ParquetDecodingException("ALP vector " + (v - 1) + " at offset " + vectorOffsets[v - 1]
+            + " spans " + span + " bytes, more than the " + maxVectorBytes + " any vector can occupy");
       }
     }
   }
 
   /**
-   * The position just past the last byte vector {@code vectorIndex} may read, derived from the
-   * next vector's offset (or the end of the body for the final vector).
+   * The most bytes one vector can occupy: its fixed headers, every value bit-packed at the full
+   * width of the type, and every value also stored as an exception. Vector sizes are multiples of
+   * the 8-value packing group, so the packed section needs no padding allowance.
+   */
+  static int maxVectorBytes(int valuesPerVector, int minVectorBytes, int valueSize) {
+    int packed = valuesPerVector * valueSize;
+    int exceptions = valuesPerVector * (Short.BYTES + valueSize);
+    return minVectorBytes + packed + exceptions;
+  }
+
+  /**
+   * The position in the vector data just past the last byte vector {@code vectorIndex} may read,
+   * taken from the next entry of the offset array; the trailing entry makes this valid for the last
+   * vector too.
    */
-  private int getVectorEndPosition(int vectorIndex) {
-    if (!isLast(vectorIndex, numVectors)) {
-      return vectorOffsets[vectorIndex + 1] - offsetArraySize;
+  int getVectorEndPosition(int vectorIndex) {
+    if (vectorIndex < 0 || vectorIndex >= numVectors) {
+      throw new IllegalArgumentException(
+          "ALP vector index " + vectorIndex + " is out of range for " + numVectors + " vectors");
     }
-    return vectorsData.limit();
+    return vectorOffsets[vectorIndex + 1] - offsetArraySize;
   }
 
   // Offsets in the page are relative to the compression body (after header),
@@ -278,14 +310,15 @@ protected final void decodeVector(int vectorIndex) {
     // Vector header, part two: ForInfo, whose width depends on the value type.
     pos = readForInfo(pos, vectorIndex, forInfo);
 
-    // Data section, part one: PackedValues.
-    pos = decodePackedValues(pos, vectorSize, exponent, factor, forInfo);
-    // The packed values' width comes from this vector's own ForInfo, so confirm they did not run
-    // past where the next vector begins; otherwise this vector would decode another one's bytes.
-    if (pos > vectorEnd) {
-      throw new ParquetDecodingException("ALP vector " + vectorIndex + " packed values end at " + pos
-          + ", past the start of the next vector at " + vectorEnd);
+    // Data section, part one: PackedValues. Their width comes from this vector's own ForInfo, so
+    // confirm they fit before unpacking; otherwise this vector would decode the next one's bytes,
+    // or for the last vector read past the end of the page.
+    int packedEnd = pos + BytesUtils.paddedByteCountFromBits(vectorSize * forInfo.bitWidth);
+    if (packedEnd > vectorEnd) {
+      throw new ParquetDecodingException("ALP vector " + vectorIndex + " packed values end at " + packedEnd
+          + ", past the end of the vector at " + vectorEnd);
     }
+    pos = decodePackedValues(pos, vectorSize, exponent, factor, forInfo);
 
     // Data section, parts two and three: ExceptionPositions then ExceptionValues.
     if (numExceptions > 0) {
@@ -362,9 +395,4 @@ static int readExceptionPositions(ByteBuffer data, int pos, int numExceptions, i
    * decoded values.
    */
   protected abstract void applyExceptions(int pos, int numExceptions, int vectorSize);
-
-  /**
-   * Reads {@code numExceptions} original values starting at {@code pos} and writes them into the
-   * decoded buffer at the positions previously read into {@link #excPositionsBuffer}.
-   */
 }
@@ -275,7 +275,7 @@ public void rejectsBitWidthTooLargeFloat() throws Exception {
   @Test
   public void rejectsPackedValuesRunningPastVectorEnd() throws Exception {
     // 16 vectors, so the inflated packed body stays inside the page and crosses into the vectors
-    // that follow rather than simply running off the end, which the buffer bounds would catch.
+    // that follow; the last-vector case, which would run off the end of the page, is below.
     byte[] page = validDoublePage(256, 16);
     int v0 = firstVectorOffset(page);
     // bitWidth lives after ALP_INFO(4) + frameOfReference(8). 64 is the maximum legal width, so it
@@ -288,6 +288,23 @@ public void rejectsPackedValuesRunningPastVectorEnd() throws Exception {
         .hasMessageContaining("packed values end at");
   }
 
+  /**
+   * The same forged bit width on the last vector would read past the end of the page. It must be
+   * rejected as a decoding error before unpacking, not surface as an IndexOutOfBoundsException.
+   */
+  @Test
+  public void rejectsPackedValuesRunningPastEndOfLastVector() throws Exception {
+    byte[] page = validDoublePage(20, 8); // 3 vectors, the last holding 4 values
+    int lastOffset = ByteBuffer.wrap(page).order(ByteOrder.LITTLE_ENDIAN).getInt(7 + 2 * Integer.BYTES);
+    page[7 + lastOffset + 12] = (byte) 64;
+    AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+    reader.initFromPage(20, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+    reader.skip(16);
+    assertThatThrownBy(reader::readDouble)
+        .isInstanceOf(ParquetDecodingException.class)
+        .hasMessageContaining("ALP vector 2 packed values end at");
+  }
+
   /**
    * An exception count that is legal against the vector size can still declare an exception block
    * larger than the vector holds. numExceptions == vectorSize passes the AlpInfo check and then
@@ -323,6 +340,40 @@ public void rejectsExceptionPositionOutOfBounds() {
         .hasMessageContaining("ALP exception position 16 out of bounds for vectorSize 16");
   }
 
+  /** Each vector must span at least its fixed headers, the last one included via the end entry. */
+  @Test
+  public void rejectsVectorShorterThanItsHeaders() {
+    // offset array of 2 entries (8 bytes), then vectors at 8 and 30, data ending at 40.
+    assertThatThrownBy(() -> AlpValuesReader.validateVectorOffsets(new int[] {8, 30, 40}, 17, 1000))
+        .isInstanceOf(ParquetDecodingException.class)
+        .hasMessageContaining("ALP vector 1 at offset 30 spans 10 bytes, fewer than the 17");
+  }
+
+  /** No vector can be longer than every value packed at full width and also stored as an exception. */
+  @Test
+  public void rejectsVectorLongerThanAnyVectorCanBe() {
+    int max = AlpValuesReader.maxVectorBytes(16, 17, Double.BYTES);
+    assertThat(max).isEqualTo(17 + 16 * 8 + 16 * (2 + 8));
+    AlpValuesReader.validateVectorOffsets(new int[] {8, 8 + max, 8 + max + 17}, 17, max);
+    assertThatThrownBy(() -> AlpValuesReader.validateVectorOffsets(new int[] {8, 9 + max, 9 + max + 17}, 17, max))
+        .isInstanceOf(ParquetDecodingException.class)
+        .hasMessageContaining("ALP vector 0 at offset 8 spans " + (max + 1) + " bytes, more than the " + max);
+  }
+
+  /** The end entry makes every in-range index valid, and anything else is a caller error. */
+  @Test
+  public void vectorEndPositionRejectsOutOfRangeIndex() throws Exception {
+    byte[] page = validDoublePage(32, 16);
+    AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+    reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+    assertThat(reader.getVectorEndPosition(1)).isEqualTo(page.length - 7 - 2 * Integer.BYTES);
+    for (int bad : new int[] {-1, 2, Integer.MAX_VALUE}) {
+      assertThatThrownBy(() -> reader.getVectorEndPosition(bad))
+          .isInstanceOf(IllegalArgumentException.class)
+          .hasMessageContaining("out of range for 2 vectors");
+    }
+  }
+
   /** The last vector is partial whenever the value count is not a whole number of vectors. */
   @Test
   public void vectorSizeAccountsForAPartialLastVector() {
 
@@ -342,16 +342,29 @@ public void testForgedVectorOffsetsAreRejected() throws Exception {
         .putInt(offsetArrayStart + Integer.BYTES, firstOffset);
     assertThatThrownBy(() -> readAllDoubles(nonIncreasing, values.length))
         .isInstanceOf(ParquetDecodingException.class)
-        .hasMessageContaining("ALP vector offsets must increase");
+        .hasMessageContaining("ALP vector 0 at offset " + firstOffset + " spans 0 bytes, fewer than");
 
-    // An offset past the end of the page body must be rejected, not read out of bounds.
+    // An offset past the end of the page body must be rejected, not read out of bounds. The
+    // trailing end-of-data entry means the vector before it is then too long or the one after it
+    // runs backwards; here vector 0 claims more bytes than any vector can occupy.
     byte[] pastEnd = valid.clone();
     ByteBuffer.wrap(pastEnd)
         .order(ByteOrder.LITTLE_ENDIAN)
         .putInt(offsetArrayStart + Integer.BYTES, valid.length * 4);
     assertThatThrownBy(() -> readAllDoubles(pastEnd, values.length))
         .isInstanceOf(ParquetDecodingException.class)
-        .hasMessageContaining("ALP vector 1 offset");
+        .hasMessageContaining("ALP vector 0 at offset " + firstOffset + " spans")
+        .hasMessageContaining("more than the");
+
+    // Pointing the last vector just past the end of the data leaves it negative room: the
+    // trailing entry turns this into a span check instead of an out-of-bounds read.
+    byte[] lastPastEnd = valid.clone();
+    int lastEntry = offsetArrayStart + 3 * Integer.BYTES;
+    int dataEnd = valid.length - offsetArrayStart;
+    ByteBuffer.wrap(lastPastEnd).order(ByteOrder.LITTLE_ENDIAN).putInt(lastEntry, dataEnd + 1);
+    assertThatThrownBy(() -> readAllDoubles(lastPastEnd, values.length))
+        .isInstanceOf(ParquetDecodingException.class)
+        .hasMessageContaining("ALP vector 3 at offset " + (dataEnd + 1) + " spans -1 bytes");
 
     // Sanity: the untouched page still decodes, so the checks above are not rejecting everything.
     assertThat(secondOffset).isGreaterThan(firstOffset);
@@ -421,6 +434,56 @@ public void testSpanBoundAcceptsEveryLegitimatePageShape() throws Exception {
     assertThat(shapes).isEqualTo(84);
   }
 
+  /** The float counterpart of the sweep above: the span bound is computed per value type. */
+  @Test
+  public void testFloatSpanBoundAcceptsEveryLegitimatePageShape() throws Exception {
+    Random rnd = new Random(42);
+    int shapes = 0;
+    for (int perVector : new int[] {8, 16, 1024, 4096}) {
+      for (int count :
+          new int[] {1, 2, perVector - 1, perVector, perVector + 1, 2 * perVector + 1, 5 * perVector + 3}) {
+        for (int profile = 0; profile < 3; profile++) {
+          float[] values = new float[count];
+          for (int i = 0; i < count; i++) {
+            if (profile == 0) {
+              values[i] = i / 8.0f;
+            } else if (profile == 1) {
+              values[i] = (i % 2 == 0) ? 2e9f : -2e9f;
+            } else {
+              values[i] = Float.intBitsToFloat(rnd.nextInt());
+            }
+          }
+          assertFloatsRoundTripExactly(values, perVector, profile);
+          shapes++;
+        }
+      }
+    }
+    assertThat(shapes).isEqualTo(84);
+  }
+
+  private void assertFloatsRoundTripExactly(float[] values, int perVector, int profile) throws Exception {
+    AlpValuesWriter.FloatAlpValuesWriter writer =
+        new AlpValuesWriter.FloatAlpValuesWriter(1 << 22, 1 << 22, new DirectByteBufferAllocator(), perVector);
+    byte[] page;
+    try {
+      for (float v : values) {
+        writer.writeFloat(v);
+      }
+      page = writer.getBytes().toByteArray();
+    } finally {
+      writer.reset();
+      writer.close();
+    }
+
+    AlpValuesReaderForFloat reader = new AlpValuesReaderForFloat();
+    reader.initFromPage(values.length, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+    for (int i = 0; i < values.length; i++) {
+      assertThat(Float.floatToRawIntBits(reader.readFloat()))
+          .as("perVector=%s count=%s profile=%s index=%s", perVector, values.length, profile, i)
+          .isEqualTo(Float.floatToRawIntBits(values[i]));
+    }
+  }
+
   private void assertDoublesRoundTripExactly(double[] values, int perVector, int profile) throws Exception {
     AlpValuesWriter.DoubleAlpValuesWriter writer =
         new AlpValuesWriter.DoubleAlpValuesWriter(1 << 22, 1 << 22, new DirectByteBufferAllocator(), perVector);