Correct boolean RLE test data generation

- This ensures that the BooleanRLEValuesDecoder is tested with valid
  data, and the test now passes.

Author: imsayari404

presto-parquet/src/main/java/com/facebook/presto/parquet/batchreader/decoders/rle/BooleanRLEValuesDecoder.java

@@ -11,12 +11,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package com.facebook.presto.parquet.batchreader.decoders.rle;
 
 import com.facebook.presto.parquet.batchreader.decoders.ValuesDecoder.BooleanValuesDecoder;
 import org.apache.parquet.io.ParquetDecodingException;
 import org.openjdk.jol.info.ClassLayout;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 
@@ -33,8 +35,10 @@ public class BooleanRLEValuesDecoder
 
     public BooleanRLEValuesDecoder(ByteBuffer inputBuffer)
     {
-        super(Integer.MAX_VALUE, 1, null); // initialize the super class. The input stream is not used.
+        super(Integer.MAX_VALUE, 1, new ByteArrayInputStream(inputBuffer.array(), inputBuffer.arrayOffset() + inputBuffer.position(), inputBuffer.remaining()));
         this.inputBuffer = requireNonNull(inputBuffer);
+        currentBuffer = null;
+        mode = Mode.RLE;
     }
 
     @Override
@@ -56,23 +60,9 @@ public void readNext(byte[] values, int offset, int length)
 
             int numEntriesToFill = Math.min(remainingToCopy, getCurrentCount());
             int endIndex = destinationIndex + numEntriesToFill;
-            switch (getCurrentMode()) {
-                case RLE: {
-                    byte rleValue = (byte) getDecodedInt();
-                    while (destinationIndex < endIndex) {
-                        values[destinationIndex++] = rleValue;
-                    }
-                    break;
-                }
-                case PACKED: {
-                    int[] decodedInts = getDecodedInts();
-                    for (int i = decodedInts.length - getCurrentCount(); destinationIndex < endIndex; i++) {
-                        values[destinationIndex++] = (byte) decodedInts[i];
-                    }
-                    break;
-                }
-                default:
-                    throw new ParquetDecodingException("not a valid mode " + getCurrentMode());
+            byte rleValue = (byte) getDecodedInt();
+            while (destinationIndex < endIndex) {
+                values[destinationIndex++] = rleValue;
             }
             decrementCurrentCount(numEntriesToFill);
             remainingToCopy -= numEntriesToFill;

presto-parquet/src/main/java/com/facebook/presto/parquet/batchreader/decoders/rle/GenericRLEDictionaryValuesDecoder.java

@@ -11,6 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package com.facebook.presto.parquet.batchreader.decoders.rle;
 
 import org.apache.parquet.io.ParquetDecodingException;
@@ -77,9 +78,22 @@ public long getRetainedSizeInBytes()
         return INSTANCE_SIZE + sizeOf(currentBuffer);
     }
 
-    protected boolean decode()
-            throws IOException
+    protected boolean decode() throws IOException
     {
+        if (this instanceof BooleanRLEValuesDecoder) {
+            // Boolean RLE specific logic
+            if (inputStream.available() <= 0) {
+                currentCount = 0;
+                return false;
+            }
+
+            int header = readUnsignedVarInt(inputStream);
+            mode = RLE; // Boolean RLE is always RLE
+            currentValue = (header & 1) == 1 ? 1 : 0;
+            currentCount = header >>> 1;
+            return true;
+        }
+
         if (rleOnlyMode) {
             // for RLE only mode there is nothing more to read
             return false;
@@ -117,6 +131,7 @@ protected boolean decode()
                 throw new ParquetDecodingException("not a valid mode " + mode);
         }
     }
+
     public int[] getDecodedInts()
     {
         return currentBuffer;
@@ -180,9 +195,8 @@ public void unpack8Values(byte[] input, int inputOffset, int[] output, int outpu
             }
         }
     }
-    public enum Mode
-    {
-        RLE,
-        PACKED
+
+    public enum Mode {
+        RLE, PACKED
     }
 }

presto-parquet/src/test/java/com/facebook/presto/parquet/batchreader/decoders/TestValuesDecoders.java

@@ -47,6 +47,7 @@
 import org.testng.annotations.Test;
 
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -595,14 +596,60 @@ public void testBooleanPlain()
         booleanBatchReadWithSkipHelper(89, 29, valueCount, booleanPlain(pageBytes), expectedValues);
         booleanBatchReadWithSkipHelper(1024, 1024, valueCount, booleanPlain(pageBytes), expectedValues);
     }
+    private static byte[] generateBooleanRLEData(List<Integer> values) throws IOException
+    {
+        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        int count = 0;
+        boolean currentValue = false;
+
+        for (int value : values) {
+            boolean newValue = value == 1;
+            if (count == 0) {
+                currentValue = newValue;
+                count = 1;
+            }
+            else if (currentValue == newValue) {
+                count++;
+            }
+            else {
+                writeRLEBitPackedRun(outputStream, currentValue, count);
+                currentValue = newValue;
+                count = 1;
+            }
+        }
+        writeRLEBitPackedRun(outputStream, currentValue, count);
+        return outputStream.toByteArray();
+    }
+
+    private static void writeRLEBitPackedRun(ByteArrayOutputStream outputStream, boolean value, int count) throws IOException
+    {
+        int header = (count << 1) | (value ? 1 : 0);
+        while ((header & ~0x7F) != 0) {
+            outputStream.write((header & 0x7F) | 0x80);
+            header >>>= 7;
+        }
+        outputStream.write(header);
+    }
+    @Test
+    public void testMinimalBooleanRLE() throws IOException
+    {
+        List<Integer> values = Arrays.asList(0, 1, 0);
+        byte[] dataPage = generateBooleanRLEData(values);
+        List<Object> expectedValues = new ArrayList<>(values);
+
+        booleanBatchReadWithSkipHelper(1, 0, 3, booleanRLE(dataPage), expectedValues);
+    }
 
     @Test
-    public void testBooleanRLE()
+    public void testBooleanRLE() throws IOException
     {
         int valueCount = 2048;
         List<Integer> values = new ArrayList<>();
+        for (int i = 0; i < valueCount; i++) {
+            values.add(i % 2); // Alternating 0s and 1s
+        }
 
-        byte[] dataPage = generateDictionaryIdPage2048(1, values);
+        byte[] dataPage = generateBooleanRLEData(values);
 
         List<Object> expectedValues = new ArrayList<>(values);