[SPARK-16962][CORE][SQL] Fix misaligned record accesses for SPARC architectures

## What changes were proposed in this pull request?

Made changes to record length offsets to make them uniform throughout various areas of Spark core and unsafe

## How was this patch tested?

This change affects only SPARC architectures and was tested on X86 architectures as well for regression.

Author: sumansomasundar <suman.somasundar@oracle.com>

Closes #14762 from sumansomasundar/master.

1 files changed, 10 insertions, 1 deletions

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/compression/CompressibleColumnBuilder.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/compression/CompressibleColumnBuilder.scala
index 0f4680e502..d1fece05a8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/compression/CompressibleColumnBuilder.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/compression/CompressibleColumnBuilder.scala
@@ -23,6 +23,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.execution.columnar.{ColumnBuilder, NativeColumnBuilder}
 import org.apache.spark.sql.types.AtomicType
+import org.apache.spark.unsafe.Platform
 
 /**
  * A stackable trait that builds optionally compressed byte buffer for a column.  Memory layout of
@@ -61,8 +62,12 @@ private[columnar] trait CompressibleColumnBuilder[T <: AtomicType]
     super.initialize(initialSize, columnName, useCompression)
   }
 
+  // The various compression schemes, while saving memory use, cause all of the data within
+  // the row to become unaligned, thus causing crashes.  Until a way of fixing the compression
+  // is found to also allow aligned accesses this must be disabled for SPARC.
+
   protected def isWorthCompressing(encoder: Encoder[T]) = {
-    encoder.compressionRatio < 0.8
+    CompressibleColumnBuilder.unaligned && encoder.compressionRatio < 0.8
   }
 
   private def gatherCompressibilityStats(row: InternalRow, ordinal: Int): Unit = {
@@ -103,3 +108,7 @@ private[columnar] trait CompressibleColumnBuilder[T <: AtomicType]
     encoder.compress(nonNullBuffer, compressedBuffer)
   }
 }
+
+private[columnar] object CompressibleColumnBuilder {
+  val unaligned = Platform.unaligned()
+}