fix endianness related issue with complex metric compression

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/hll/HllSketchHolderObjectStrategy.java

@@ -79,6 +79,12 @@ public HllSketchHolder fromByteBufferSafe(ByteBuffer buffer, int numBytes)
     );
   }
 
+  @Override
+  public ByteOrder getByteOrder()
+  {
+    return ByteOrder.LITTLE_ENDIAN;
+  }
+
   /**
    * Checks if a sketch is empty and can be converted to null. Returns true if it is and false if it is not, or if is
    * not possible to say for sure.

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/kll/KllDoublesSketchObjectStrategy.java

@@ -73,4 +73,10 @@ public KllDoublesSketch fromByteBufferSafe(ByteBuffer buffer, int numBytes)
         SafeWritableMemory.wrap(buffer, ByteOrder.LITTLE_ENDIAN).region(buffer.position(), numBytes)
     );
   }
+
+  @Override
+  public ByteOrder getByteOrder()
+  {
+    return ByteOrder.LITTLE_ENDIAN;
+  }
 }

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/kll/KllFloatsSketchObjectStrategy.java

@@ -73,4 +73,10 @@ public KllFloatsSketch fromByteBufferSafe(ByteBuffer buffer, int numBytes)
         SafeWritableMemory.wrap(buffer, ByteOrder.LITTLE_ENDIAN).region(buffer.position(), numBytes)
     );
   }
+
+  @Override
+  public ByteOrder getByteOrder()
+  {
+    return ByteOrder.LITTLE_ENDIAN;
+  }
 }

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/quantiles/DoublesSketchObjectStrategy.java

@@ -73,4 +73,10 @@ public DoublesSketch fromByteBufferSafe(ByteBuffer buffer, int numBytes)
         SafeWritableMemory.wrap(buffer, ByteOrder.LITTLE_ENDIAN).region(buffer.position(), numBytes)
     );
   }
+
+  @Override
+  public ByteOrder getByteOrder()
+  {
+    return ByteOrder.LITTLE_ENDIAN;
+  }
 }

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/theta/SketchHolderObjectStrategy.java

@@ -80,4 +80,10 @@ public SketchHolder fromByteBufferSafe(ByteBuffer buffer, int numBytes)
         SafeWritableMemory.wrap(buffer, ByteOrder.LITTLE_ENDIAN).region(buffer.position(), numBytes)
     );
   }
+
+  @Override
+  public ByteOrder getByteOrder()
+  {
+    return ByteOrder.LITTLE_ENDIAN;
+  }
 }

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/tuple/ArrayOfDoublesSketchObjectStrategy.java

@@ -72,4 +72,10 @@ public ArrayOfDoublesSketch fromByteBufferSafe(ByteBuffer buffer, int numBytes)
         SafeWritableMemory.wrap(buffer, ByteOrder.LITTLE_ENDIAN).region(buffer.position(), numBytes)
     );
   }
+
+  @Override
+  public ByteOrder getByteOrder()
+  {
+    return ByteOrder.LITTLE_ENDIAN;
+  }
 }

processing/src/main/java/org/apache/druid/segment/data/CompressedBlockReader.java

@@ -59,6 +59,7 @@ public final class CompressedBlockReader implements Closeable
   public static Supplier<CompressedBlockReader> fromByteBuffer(
       ByteBuffer buffer,
       ByteOrder byteOrder,
+      ByteOrder valueOrder,
       boolean copyValuesOnRead
   )
   {
@@ -94,7 +95,8 @@ public static Supplier<CompressedBlockReader> fromByteBuffer(
           copyValuesOnRead,
           offsetView.asReadOnlyBuffer(),
           compressedDataView.asReadOnlyBuffer().order(byteOrder),
-          byteOrder
+          byteOrder,
+          valueOrder
       );
     }
     throw new IAE("Unknown version[%s]", versionFromBuffer);
@@ -123,7 +125,8 @@ public CompressedBlockReader(
       boolean copyValuesOnRead,
       IntBuffer endOffsetsBuffer,
       ByteBuffer compressedDataBuffer,
-      ByteOrder byteOrder
+      ByteOrder compressionOrder,
+      ByteOrder valueOrder
   )
   {
     this.decompressor = compressionStrategy.getDecompressor();
@@ -134,11 +137,11 @@ public CompressedBlockReader(
     this.endOffsetsBuffer = endOffsetsBuffer;
     this.compressedDataBuffer = compressedDataBuffer;
     this.closer = Closer.create();
-    this.decompressedDataBufferHolder = CompressedPools.getByteBuf(byteOrder);
+    this.decompressedDataBufferHolder = CompressedPools.getByteBuf(compressionOrder);
     closer.register(decompressedDataBufferHolder);
     this.decompressedDataBuffer = decompressedDataBufferHolder.get();
     this.decompressedDataBuffer.clear();
-    this.byteOrder = byteOrder;
+    this.byteOrder = valueOrder;
   }
 
   /**

processing/src/main/java/org/apache/druid/segment/data/CompressedLongsReader.java

@@ -29,7 +29,7 @@ public final class CompressedLongsReader implements ColumnarLongs
 {
   public static Supplier<CompressedLongsReader> fromByteBuffer(ByteBuffer buffer, ByteOrder order)
   {
-    final Supplier<CompressedBlockReader> baseReader = CompressedBlockReader.fromByteBuffer(buffer, order, false);
+    final Supplier<CompressedBlockReader> baseReader = CompressedBlockReader.fromByteBuffer(buffer, order, order, false);
     return () -> new CompressedLongsReader(baseReader.get());
   }
 

processing/src/main/java/org/apache/druid/segment/data/CompressedVariableSizedBlobColumnSupplier.java

@@ -34,17 +34,29 @@ public class CompressedVariableSizedBlobColumnSupplier implements Supplier<Compr
   public static CompressedVariableSizedBlobColumnSupplier fromByteBuffer(
       String filenameBase,
       ByteBuffer buffer,
-      ByteOrder order,
+      ByteOrder compressionOrder,
       SmooshedFileMapper mapper
   ) throws IOException
   {
-    return fromByteBuffer(filenameBase, buffer, order, false, mapper);
+    return fromByteBuffer(filenameBase, buffer, compressionOrder, compressionOrder, false, mapper);
   }
 
   public static CompressedVariableSizedBlobColumnSupplier fromByteBuffer(
       String filenameBase,
       ByteBuffer buffer,
-      ByteOrder order,
+      ByteOrder compressionOrder,
+      ByteOrder valueOrder,
+      SmooshedFileMapper mapper
+  ) throws IOException
+  {
+    return fromByteBuffer(filenameBase, buffer, compressionOrder, valueOrder, false, mapper);
+  }
+
+  public static CompressedVariableSizedBlobColumnSupplier fromByteBuffer(
+      String filenameBase,
+      ByteBuffer buffer,
+      ByteOrder compressionOrder,
+      ByteOrder valueOrder,
       boolean copyValuesOnRead,
       SmooshedFileMapper mapper
   ) throws IOException
@@ -59,7 +71,14 @@ public static CompressedVariableSizedBlobColumnSupplier fromByteBuffer(
       final ByteBuffer dataBuffer = mapper.mapFile(
           CompressedVariableSizedBlobColumnSerializer.getCompressedBlobsFileName(filenameBase)
       );
-      return new CompressedVariableSizedBlobColumnSupplier(offsetsBuffer, dataBuffer, order, numElements, copyValuesOnRead);
+      return new CompressedVariableSizedBlobColumnSupplier(
+          offsetsBuffer,
+          dataBuffer,
+          compressionOrder,
+          valueOrder,
+          numElements,
+          copyValuesOnRead
+      );
     }
     throw new IAE("Unknown version[%s]", versionFromBuffer);
   }
@@ -72,14 +91,20 @@ public static CompressedVariableSizedBlobColumnSupplier fromByteBuffer(
   private CompressedVariableSizedBlobColumnSupplier(
       ByteBuffer offsetsBuffer,
       ByteBuffer dataBuffer,
-      ByteOrder order,
+      ByteOrder compressionOrder,
+      ByteOrder valueOrder,
       int numElements,
       boolean copyValuesOnRead
   )
   {
     this.numElements = numElements;
-    this.offsetReaderSupplier = CompressedLongsReader.fromByteBuffer(offsetsBuffer, order);
-    this.blockDataReaderSupplier = CompressedBlockReader.fromByteBuffer(dataBuffer, order, copyValuesOnRead);
+    this.offsetReaderSupplier = CompressedLongsReader.fromByteBuffer(offsetsBuffer, compressionOrder);
+    this.blockDataReaderSupplier = CompressedBlockReader.fromByteBuffer(
+        dataBuffer,
+        compressionOrder,
+        valueOrder,
+        copyValuesOnRead
+    );
   }
 
   @Override

processing/src/main/java/org/apache/druid/segment/data/ObjectStrategy.java

@@ -26,6 +26,7 @@
 import javax.annotation.Nullable;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.Comparator;
 
 @ExtensionPoint
@@ -77,6 +78,18 @@ default boolean readRetainsBufferReference()
     return true;
   }
 
+  /**
+   * Preferred byte order to read values from a buffer with {@link #fromByteBuffer(ByteBuffer, int)} and similar methods
+   *
+   * This is currently only used as a helper for {@link org.apache.druid.segment.serde.CompressedComplexColumn} when
+   * {@link org.apache.druid.segment.IndexSpec#complexMetricCompression} is set, so that the buffer does not need
+   * to be explicitly set inside the object strategy when materializing values
+   */
+  default ByteOrder getByteOrder()
+  {
+    return ByteOrder.BIG_ENDIAN;
+  }
+
   /**
    * Reads 4-bytes numBytes from the given buffer, and then delegates to {@link #fromByteBuffer(ByteBuffer, int)}.
    */

processing/src/main/java/org/apache/druid/segment/nested/NestedDataComplexTypeSerde.java

@@ -43,6 +43,7 @@
 import javax.annotation.Nullable;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 
 public class NestedDataComplexTypeSerde extends ComplexMetricSerde
 {
@@ -147,6 +148,12 @@ public boolean readRetainsBufferReference()
       {
         return false;
       }
+
+      @Override
+      public ByteOrder getByteOrder()
+      {
+        return ByteOrder.LITTLE_ENDIAN;
+      }
     };
   }
 

processing/src/main/java/org/apache/druid/segment/serde/CompressedComplexColumnSupplier.java

@@ -67,6 +67,7 @@ public static CompressedComplexColumnSupplier read(
             ),
             fileBuffer,
             metadata.getByteOrder(),
+            objectStrategy.getByteOrder(),
             objectStrategy.readRetainsBufferReference(),
             mapper
         );