diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/Statistics.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/Statistics.scala
index 43455c989c0f4..f3e2147b8f974 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/Statistics.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/Statistics.scala
@@ -98,7 +98,7 @@ case class StringColumnStat(statRow: InternalRow) {
   // The indices here must be consistent with `ColumnStatStruct.stringColumnStat`.
   val numNulls: Long = statRow.getLong(0)
   val avgColLen: Double = statRow.getDouble(1)
-  val maxColLen: Long = statRow.getLong(2)
+  val maxColLen: Long = statRow.getInt(2)
   val ndv: Long = statRow.getLong(3)
 }
 
@@ -106,7 +106,7 @@ case class BinaryColumnStat(statRow: InternalRow) {
   // The indices here must be consistent with `ColumnStatStruct.binaryColumnStat`.
   val numNulls: Long = statRow.getLong(0)
   val avgColLen: Double = statRow.getDouble(1)
-  val maxColLen: Long = statRow.getLong(2)
+  val maxColLen: Long = statRow.getInt(2)
 }
 
 case class BooleanColumnStat(statRow: InternalRow) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsColumnSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsColumnSuite.scala
index 0ee0547c45591..f1a201abd8da6 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsColumnSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsColumnSuite.scala
@@ -150,7 +150,7 @@ class StatisticsColumnSuite extends StatisticsTest {
       val colStat = ColumnStat(InternalRow(
         values.count(_.isEmpty).toLong,
         nonNullValues.map(_.length).sum / nonNullValues.length.toDouble,
-        nonNullValues.map(_.length).max.toLong,
+        nonNullValues.map(_.length).max.toInt,
         nonNullValues.distinct.length.toLong))
       (f, colStat)
     }
@@ -165,7 +165,7 @@ class StatisticsColumnSuite extends StatisticsTest {
       val colStat = ColumnStat(InternalRow(
         values.count(_.isEmpty).toLong,
         nonNullValues.map(_.length).sum / nonNullValues.length.toDouble,
-        nonNullValues.map(_.length).max.toLong))
+        nonNullValues.map(_.length).max.toInt))
       (f, colStat)
     }
     checkColStats(df, expectedColStatsSeq)
@@ -255,10 +255,10 @@ class StatisticsColumnSuite extends StatisticsTest {
               doubleSeq.distinct.length.toLong))
         case StringType =>
           ColumnStat(InternalRow(0L, stringSeq.map(_.length).sum / stringSeq.length.toDouble,
-                stringSeq.map(_.length).max.toLong, stringSeq.distinct.length.toLong))
+                stringSeq.map(_.length).max.toInt, stringSeq.distinct.length.toLong))
         case BinaryType =>
           ColumnStat(InternalRow(0L, binarySeq.map(_.length).sum / binarySeq.length.toDouble,
-                binarySeq.map(_.length).max.toLong))
+                binarySeq.map(_.length).max.toInt))
         case BooleanType =>
           ColumnStat(InternalRow(0L, booleanSeq.count(_.equals(true)).toLong,
               booleanSeq.count(_.equals(false)).toLong))
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/StatisticsSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/StatisticsSuite.scala
index 99dd080683d40..85228bb00123d 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/StatisticsSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/StatisticsSuite.scala
@@ -378,7 +378,7 @@ class StatisticsSuite extends QueryTest with TestHiveSingleton with SQLTestUtils
             ColumnStat(InternalRow(0L, intSeq.max, intSeq.min, intSeq.distinct.length.toLong))
           case StringType =>
             ColumnStat(InternalRow(0L, stringSeq.map(_.length).sum / stringSeq.length.toDouble,
-              stringSeq.map(_.length).max.toLong, stringSeq.distinct.length.toLong))
+              stringSeq.map(_.length).max.toInt, stringSeq.distinct.length.toLong))
           case BooleanType =>
             ColumnStat(InternalRow(0L, booleanSeq.count(_.equals(true)).toLong,
               booleanSeq.count(_.equals(false)).toLong))