[Issue #531] support configurable endianness (#534)

The endianness of Pixels writer is configured by column.chunk.little.endian=true/false.
The endianness is then saved in the ColumnChunkIndex of each column chunk.
The Pixels column readers will check the ColumnChunkIndex to use the right endianness.
This is currently implemented in the Java codebase, and the C++ codebase should at least check and report errors for unsupported endianness.

Author: bianhq

pixels-cache/src/main/java/io/pixelsdb/pixels/cache/PixelsPhysicalReader.java

@@ -25,6 +25,7 @@
 import io.pixelsdb.pixels.core.PixelsProto;
 
 import java.io.IOException;
+import java.nio.ByteOrder;
 
 /**
  * @author guodong
@@ -48,7 +49,7 @@ private PixelsProto.FileTail readFileTail()
             {
                 long fileLen = physicalReader.getFileLength();
                 physicalReader.seek(fileLen - Long.BYTES);
-                long fileTailOffset = physicalReader.readLong();
+                long fileTailOffset = physicalReader.readLong(ByteOrder.BIG_ENDIAN);
                 int fileTailLength = (int) (fileLen - fileTailOffset - Long.BYTES);
                 physicalReader.seek(fileTailOffset);
                 byte[] fileTailBuffer = new byte[fileTailLength];

pixels-common/src/main/java/io/pixelsdb/pixels/common/physical/PhysicalReader.java

@@ -22,6 +22,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -33,8 +34,21 @@ public interface PhysicalReader extends Closeable
 {
     long getFileLength() throws IOException;
 
+    /**
+     * Set the current offset to the desired value.
+     * @param desired the desired offset
+     * @throws IOException
+     */
     void seek(long desired) throws IOException;
 
+    /**
+     * Read a byte buffer of the given length from the current offset. The byte order
+     * of the byte buffer is default byte order of {@link ByteBuffer}. Please check and
+     * reset it if necessary.
+     * @param length the number of bytes to read
+     * @return the byte buffer been read
+     * @throws IOException
+     */
     ByteBuffer readFully(int length) throws IOException;
 
     void readFully(byte[] buffer) throws IOException;
@@ -61,9 +75,21 @@ default CompletableFuture<ByteBuffer> readAsync(long offset, int length) throws
         throw new UnsupportedOperationException("asynchronous read is not supported for " + getStorageScheme().name());
     }
 
-    long readLong() throws IOException;
+    /**
+     * Read an eight-byte signed integer from the current offset using the specified byte order.
+     * @param byteOrder the byte order
+     * @return the integer been read
+     * @throws IOException
+     */
+    long readLong(ByteOrder byteOrder) throws IOException;
 
-    int readInt() throws IOException;
+    /**
+     * Read a four-byte signed integer from the current offset using the specified byte order.
+     * @param byteOrder the byte order
+     * @return the integer been read
+     * @throws IOException
+     */
+    int readInt(ByteOrder byteOrder) throws IOException;
 
     void close() throws IOException;
 
@@ -78,7 +104,7 @@ default CompletableFuture<ByteBuffer> readAsync(long offset, int length) throws
     /**
      * For a file or object in the storage, it may have one or more
      * blocks. Each block has its unique id. This method returns the
-     * block id of the current block that is been reading.
+     * block id of the current block that is being reading.
      *
      * For local fs, each file has only one block id, which is also
      * the file id.

pixels-common/src/main/resources/pixels.properties

@@ -54,21 +54,29 @@ cache.read.direct=false
 
 ###### storage engine settings ######
 
-### pixels reader and writer settings ###
-# properties for pixels writer
+### pixels reader, writer, and compactor settings ###
+# the pixels stride (number of values in a pixel) for pixels writer
 pixel.stride=10000
+# the row group size in bytes for pixels writer
 row.group.size=268435456
-# The alignment is for SIMD and its unit is byte
+# the chunk alignment for pixels writer, it is for SIMD and its unit is byte
 column.chunk.alignment=32
+# whether column chunks are encoded in pixels writer
 column.chunk.encoding=true
+# the little-endian is used on the column chunks in pixels writer
+column.chunk.little.endian=true
+# the block size for block-wise storage systems such as HDFS
 block.size=2147483648
+# the number of replications of each block for block-wise storage systems such as HDFS
 block.replication=1
+# for block-wise storage systems, whether padding the leftover space in current block
 block.padding=true
-compression.block.size=1
+# the number of bytes to be compressed as a block using heavy compression algorithms
+compression.block.size=1048576
+# for pixels compactor how many row groups are compacted into one file
+compact.factor=32
 # row batch size for pixels record reader, default value is 10000
 row.batch.size=10000
-# how many row groups are compacted into one file
-compact.factor=32
 
 ### file storage and I/O ###
 # the scheme of the storage systems that are enabled, e.g., hdfs,file,s3,gcs,minio,redis

pixels-core/src/main/java/io/pixelsdb/pixels/core/PixelsReaderImpl.java

@@ -38,6 +38,7 @@
 import javax.annotation.concurrent.NotThreadSafe;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Random;
@@ -171,7 +172,7 @@ public PixelsReader build()
                 // get FileTail
                 long fileLen = fsReader.getFileLength();
                 fsReader.seek(fileLen - Long.BYTES);
-                long fileTailOffset = fsReader.readLong();
+                long fileTailOffset = fsReader.readLong(ByteOrder.BIG_ENDIAN);
                 int fileTailLength = (int) (fileLen - fileTailOffset - Long.BYTES);
                 fsReader.seek(fileTailOffset);
                 ByteBuffer fileTailBuffer = fsReader.readFully(fileTailLength);

pixels-core/src/main/java/io/pixelsdb/pixels/core/PixelsWriter.java

@@ -29,8 +29,7 @@
  *
  * @author guodong
  */
-public interface PixelsWriter
-        extends Closeable
+public interface PixelsWriter extends Closeable
 {
     /**
      * Add row batch into the file that is not hash partitioned.

pixels-core/src/main/java/io/pixelsdb/pixels/core/PixelsWriterImpl.java

@@ -38,6 +38,7 @@
 import javax.annotation.concurrent.NotThreadSafe;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Optional;
@@ -68,6 +69,22 @@ public class PixelsWriterImpl implements PixelsWriter
 {
     private static final Logger LOGGER = LogManager.getLogger(PixelsWriterImpl.class);
 
+    static final ByteOrder WRITER_ENDIAN;
+
+    static
+    {
+        boolean littleEndian = Boolean.parseBoolean(
+                ConfigFactory.Instance().getProperty("column.chunk.little.endian"));
+        if (littleEndian)
+        {
+            WRITER_ENDIAN = ByteOrder.LITTLE_ENDIAN;
+        }
+        else
+        {
+            WRITER_ENDIAN = ByteOrder.BIG_ENDIAN;
+        }
+    }
+
     private final TypeDescription schema;
     private final int pixelStride;
     private final int rowGroupSize;
@@ -143,7 +160,7 @@ private PixelsWriterImpl(
         fileColStatRecorders = new StatsRecorder[children.size()];
         for (int i = 0; i < children.size(); ++i)
         {
-            columnWriters[i] = newColumnWriter(children.get(i), pixelStride, encoding);
+            columnWriters[i] = newColumnWriter(children.get(i), pixelStride, encoding, WRITER_ENDIAN);
             fileColStatRecorders[i] = StatsRecorder.create(children.get(i));
         }
 
@@ -610,7 +627,7 @@ private void writeRowGroup() throws IOException
              * We temporarily fix this problem by creating a new column writer for each row group.
              */
             // writer.reset();
-            columnWriters[i] = newColumnWriter(children.get(i), pixelStride, encoding);
+            columnWriters[i] = newColumnWriter(children.get(i), pixelStride, encoding, WRITER_ENDIAN);
         }
 
         // put curRowGroupIndex into rowGroupFooter

pixels-core/src/main/java/io/pixelsdb/pixels/core/compactor/PixelsCompactor.java

@@ -34,6 +34,7 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.TimeZone;
@@ -250,7 +251,7 @@ public PixelsCompactor build()
                 // get FileTail
                 long fileLen = fsReader.getFileLength();
                 fsReader.seek(fileLen - Long.BYTES);
-                long fileTailOffset = fsReader.readLong();
+                long fileTailOffset = fsReader.readLong(ByteOrder.BIG_ENDIAN);
                 int fileTailLength = (int) (fileLen - fileTailOffset - Long.BYTES);
                 fsReader.seek(fileTailOffset);
                 byte[] fileTailBuffer = new byte[fileTailLength];

pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/Dictionary.java

@@ -67,7 +67,8 @@ public interface Dictionary
      * This method prepares a {@link VisitorContext} instance for each item (key) in this dictionary,
      * and calls {@link Visitor#visit(VisitorContext)} to visit the item.
      * <p>
-     * The items <b>MUST</b> be visited in the ascending order of the item's key position, i.e., the encoded id of the item in this dictionary.
+     * The items <b>MUST</b> be visited in the ascending order of the item's key position, i.e.,
+     * the encoded id of the item in this dictionary.
      * The visitor can write (serialize) the dictionary item to an output stream.
      * </p>
      * @param visitor the visitor that is going to serialize the dictionary to an output stream.

pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/Encoder.java

@@ -49,38 +49,32 @@ public byte[] encode(char[] values, int offset, int length)
         throw new PixelsEncodingException("Encoding char values is not supported");
     }
 
-    public byte[] encode(byte[] values)
-            throws IOException
+    public byte[] encode(byte[] values) throws IOException
     {
         throw new PixelsEncodingException("Encoding byte values is not supported");
     }
 
-    public byte[] encode(byte[] values, int offset, int length)
-            throws IOException
+    public byte[] encode(byte[] values, int offset, int length) throws IOException
     {
         throw new PixelsEncodingException("Encoding byte values is not supported");
     }
 
-    public byte[] encode(long[] values)
-            throws IOException
+    public byte[] encode(long[] values) throws IOException
     {
         throw new PixelsEncodingException("Encoding long values is not supported");
     }
 
-    public byte[] encode(long[] values, int offset, int length)
-            throws IOException
+    public byte[] encode(long[] values, int offset, int length) throws IOException
     {
         throw new PixelsEncodingException("Encoding long values is not supported");
     }
 
-    public byte[] encode(int[] values)
-            throws IOException
+    public byte[] encode(int[] values) throws IOException
     {
         throw new PixelsEncodingException("Encoding int values is not supported");
     }
 
-    public byte[] encode(int[] values, int offset, int length)
-            throws IOException
+    public byte[] encode(int[] values, int offset, int length) throws IOException
     {
         throw new PixelsEncodingException("Encoding int values is not supported");
     }
@@ -100,6 +94,5 @@ public byte[] encode(double[] values)
         throw new PixelsEncodingException("Encoding double values is not supported");
     }
 
-    abstract public void close()
-            throws IOException;
+    abstract public void close() throws IOException;
 }

pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/IntDecoder.java

@@ -26,6 +26,5 @@
  */
 public abstract class IntDecoder extends Decoder
 {
-    public abstract long next()
-            throws IOException;
+    public abstract long next() throws IOException;
 }