diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java
index 1edee60bc5644..209524ba2cb42 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java
@@ -22,6 +22,7 @@
 import org.apache.parquet.bytes.ByteBufferInputStream;
 import org.apache.parquet.column.values.RequiresPreviousReader;
 import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
 import org.apache.parquet.io.api.Binary;
 import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector;
 import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
@@ -71,6 +72,24 @@ public Binary readBinary(int len) {
     return Binary.fromConstantByteArray(binaryValVector.getBinary(0));
   }
 
+  /**
+   * The prefix length is read from the file, so validate it against the previous value before
+   * using it to copy bytes out of that value. A corrupt page must fail the read instead of
+   * producing a malformed value.
+   */
+  private void checkPrefixLength(int prefixLength) {
+    if (prefixLength < 0) {
+      throw new ParquetDecodingException(
+        "Corrupted DELTA_BYTE_ARRAY page: negative prefix length: " + prefixLength);
+    }
+    int previousLength = previous == null ? 0 : previous.remaining();
+    if (prefixLength > previousLength) {
+      throw new ParquetDecodingException(
+        "Corrupted DELTA_BYTE_ARRAY page: prefix length " + prefixLength
+          + " is larger than the previous value's length " + previousLength);
+    }
+  }
+
   private void readValues(int total, WritableColumnVector c, int rowId) {
     for (int i = 0; i < total; i++) {
       // NOTE: due to PARQUET-246, it is important that we
@@ -79,6 +98,7 @@ private void readValues(int total, WritableColumnVector c, int rowId) {
       // value of the page should have an empty prefix, it may not
       // because of PARQUET-246.
       int prefixLength = prefixLengthVector.getInt(currentRow);
+      checkPrefixLength(prefixLength);
       ByteBuffer suffix = suffixReader.getBytes(currentRow);
       byte[] suffixArray = suffix.array();
       int suffixLength = suffix.limit() - suffix.position();
@@ -120,6 +140,7 @@ private void readGeoData(int total, WritableColumnVector c, int rowId, int srid,
      WKBConverterStrategy converter) {
     for (int i = 0; i < total; i++) {
       int prefixLength = prefixLengthVector.getInt(currentRow);
+      checkPrefixLength(prefixLength);
       ByteBuffer suffix = suffixReader.getBytes(currentRow);
       int suffixLength = suffix.limit() - suffix.position();
       int length = prefixLength + suffixLength;
@@ -165,6 +186,7 @@ public void skipBinary(int total) {
 
     for (int i = 0; i < total; i++) {
       int prefixLength = prefixLengthVector.getInt(currentRow);
+      checkPrefixLength(prefixLength);
       ByteBuffer suffix = suffixReader.getBytes(currentRow);
       byte[] suffixArray = suffix.array();
       int suffixLength = suffix.limit() - suffix.position();
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaByteArrayEncodingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaByteArrayEncodingSuite.scala
index d62ef9c3f197d..77a2caaa8e197 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaByteArrayEncodingSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaByteArrayEncodingSuite.scala
@@ -16,9 +16,13 @@
  */
 package org.apache.spark.sql.execution.datasources.parquet
 
-import org.apache.parquet.bytes.DirectByteBufferAllocator
+import org.apache.parquet.bytes.{ByteBufferInputStream, BytesInput, DirectByteBufferAllocator}
 import org.apache.parquet.column.values.Utils
+import org.apache.parquet.column.values.delta.DeltaBinaryPackingValuesWriterForInteger
+import org.apache.parquet.column.values.deltalengthbytearray.DeltaLengthByteArrayValuesWriter
 import org.apache.parquet.column.values.deltastrings.DeltaByteArrayWriter
+import org.apache.parquet.io.ParquetDecodingException
+import org.apache.parquet.io.api.Binary
 
 import org.apache.spark.sql.catalyst.util.STUtils
 import org.apache.spark.sql.execution.vectorized.{OnHeapColumnVector, WritableColumnVector}
@@ -59,6 +63,37 @@ class ParquetDeltaByteArrayEncodingSuite extends ParquetCompatibilityTest with S
     assertReadWriteWithSkipN(writer, reader, randvalues)
   }
 
+  test("prefix length larger than the previous value is rejected") {
+    // Craft a page by hand: a benign 2-byte first value, then a value whose prefix length
+    // claims 65536 bytes of a 2-byte previous value.
+    val is = craftPage(prefixLengths = Array(0, 65536), suffixes = Array("ab", ""))
+    reader.initFromPage(2, is)
+    writableColumnVector = new OnHeapColumnVector(2, StringType)
+    val e = intercept[ParquetDecodingException] {
+      reader.readBinary(2, writableColumnVector, 0)
+    }
+    assert(e.getMessage.contains("prefix length 65536"))
+  }
+
+  test("negative prefix length is rejected") {
+    val is = craftPage(prefixLengths = Array(0, -1), suffixes = Array("ab", "cd"))
+    reader.initFromPage(2, is)
+    writableColumnVector = new OnHeapColumnVector(2, StringType)
+    val e = intercept[ParquetDecodingException] {
+      reader.readBinary(2, writableColumnVector, 0)
+    }
+    assert(e.getMessage.contains("negative prefix length"))
+  }
+
+  test("prefix length larger than the previous value is rejected when skipping") {
+    val is = craftPage(prefixLengths = Array(0, 65536), suffixes = Array("ab", ""))
+    reader.initFromPage(2, is)
+    val e = intercept[ParquetDecodingException] {
+      reader.skipBinary(2)
+    }
+    assert(e.getMessage.contains("prefix length 65536"))
+  }
+
   test("test lengths") {
     var reader = new VectorizedDeltaBinaryPackedReader
     Utils.writeData(writer, values)
@@ -100,6 +135,19 @@ class ParquetDeltaByteArrayEncodingSuite extends ParquetCompatibilityTest with S
       geoType)
   }
 
+  /** Builds a raw DELTA_BYTE_ARRAY page from explicit prefix lengths and suffixes. */
+  private def craftPage(
+      prefixLengths: Array[Int],
+      suffixes: Array[String]): ByteBufferInputStream = {
+    val allocator = new DirectByteBufferAllocator
+    val prefixWriter =
+      new DeltaBinaryPackingValuesWriterForInteger(128, 4, 64 * 1024, 64 * 1024, allocator)
+    val suffixWriter = new DeltaLengthByteArrayValuesWriter(64 * 1024, 64 * 1024, allocator)
+    prefixLengths.foreach(prefixWriter.writeInteger)
+    suffixes.foreach(s => suffixWriter.writeBytes(Binary.fromString(s)))
+    BytesInput.concat(prefixWriter.getBytes, suffixWriter.getBytes).toInputStream
+  }
+
   private def assertGeoReadWrite(
       writer: DeltaByteArrayWriter,
       reader: VectorizedDeltaByteArrayReader,