[Issue #521] support chunk aligned compact file (#532)

Also fix an argument checking bug in PixelsCompactor.

Author: bianhq

pixels-common/src/main/java/io/pixelsdb/pixels/common/layout/InvertedSplitsIndex.java

@@ -55,7 +55,7 @@ public InvertedSplitsIndex(long version, List<String> columnOrder, List<SplitPat
             BitSet bitMap = new BitSet(this.querySplitPatterns.size());
             for (int i = 0; i < this.querySplitPatterns.size(); ++i)
             {
-                if (this.querySplitPatterns.get(i).contaiansColumn(column))
+                if (this.querySplitPatterns.get(i).containsColumn(column))
                 {
                     bitMap.set(i, true);
                 }

pixels-common/src/main/java/io/pixelsdb/pixels/common/layout/SplitPattern.java

@@ -66,7 +66,7 @@ public int getSplitSize()
         return splitSize;
     }
 
-    public boolean contaiansColumn(String column)
+    public boolean containsColumn(String column)
     {
         return this.columnSet.contains(column);
     }

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

@@ -60,10 +60,10 @@ pixel.stride=10000
 row.group.size=268435456
 # The alignment is for SIMD and its unit is byte
 column.chunk.alignment=32
+column.chunk.encoding=true
 block.size=2147483648
 block.replication=1
 block.padding=true
-encoding=true
 compression.block.size=1
 # row batch size for pixels record reader, default value is 10000
 row.batch.size=10000

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

@@ -77,6 +77,14 @@ public class PixelsWriterImpl implements PixelsWriter
     private final boolean encoding;
     private final boolean partitioned;
     private final Optional<List<Integer>> partKeyColumnIds;
+    /**
+     * The number of bytes that each column chunk is aligned to.
+     */
+    private final int chunkAlignment;
+    /**
+     * The byte buffer padded to each column chunk for alignment.
+     */
+    private final byte[] chunkPaddingBuffer;
 
     private final ColumnWriter[] columnWriters;
     private final StatsRecorder[] fileColStatRecorders;
@@ -101,14 +109,6 @@ public class PixelsWriterImpl implements PixelsWriter
     private final List<TypeDescription> children;
 
     private final ExecutorService columnWriterService = Executors.newCachedThreadPool();
-    /**
-     * The number of bytes that each column chunk is aligned to.
-     */
-    private final int chunkAlignment;
-    /**
-     * The byte buffer padded to each column chunk for alignment.
-     */
-    private final byte[] chunkPaddingBuffer;
 
     private PixelsWriterImpl(
             TypeDescription schema,
@@ -282,8 +282,7 @@ public Builder setPartKeyColumnIds(List<Integer> partitionColumnIds)
             return this;
         }
 
-        public PixelsWriter build()
-                throws PixelsWriterException
+        public PixelsWriter build() throws PixelsWriterException
         {
             requireNonNull(this.builderStorage, "storage is not set");
             requireNonNull(this.builderFilePath, "file path is not set");
@@ -399,8 +398,7 @@ public boolean isPartitioned()
     }
 
     @Override
-    public boolean addRowBatch(VectorizedRowBatch rowBatch)
-            throws IOException
+    public boolean addRowBatch(VectorizedRowBatch rowBatch) throws IOException
     {
         checkArgument(!partitioned, "this file is hash partitioned, " +
                 "use addRowBatch(rowBatch, hashValue) instead");
@@ -498,8 +496,7 @@ public void close()
         }
     }
 
-    private void writeRowGroup()
-            throws IOException
+    private void writeRowGroup() throws IOException
     {
         int rowGroupDataLength = 0;
 
@@ -660,8 +657,7 @@ private void writeRowGroup()
         this.fileContentLength += rowGroupDataLength;
     }
 
-    private void writeFileTail()
-            throws IOException
+    private void writeFileTail() throws IOException
     {
         PixelsProto.Footer footer;
         PixelsProto.PostScript postScript;
@@ -694,6 +690,7 @@ private void writeFileTail()
                 .setPixelStride(pixelStride)
                 .setWriterTimezone(timeZone.getDisplayName())
                 .setPartitioned(partitioned)
+                .setColumnChunkAlignment(chunkAlignment)
                 .setMagic(Constants.MAGIC)
                 .build();
 

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

@@ -21,6 +21,7 @@
 
 import com.google.common.collect.ImmutableList;
 import io.pixelsdb.pixels.common.physical.*;
+import io.pixelsdb.pixels.common.utils.ConfigFactory;
 import io.pixelsdb.pixels.common.utils.Constants;
 import io.pixelsdb.pixels.core.PixelsProto;
 import io.pixelsdb.pixels.core.PixelsVersion;
@@ -58,6 +59,14 @@ public class PixelsCompactor
     private final TimeZone timeZone;
     private final long fileContentLength;
     private final int fileRowNum;
+    /**
+     * The number of bytes that each column chunk is aligned to.
+     */
+    private final int chunkAlignment;
+    /**
+     * The byte buffer padded to each column chunk for alignment.
+     */
+    private final byte[] chunkPaddingBuffer;
 
     private final Storage inputStorage;
     private final PhysicalWriter fsWriter;
@@ -93,6 +102,9 @@ private PixelsCompactor(
         checkArgument(compressionBlockSize > 0, "compression block size is not positive");
         this.compressionBlockSize = compressionBlockSize;
         this.timeZone = requireNonNull(timeZone);
+        this.chunkAlignment = Integer.parseInt(ConfigFactory.Instance().getProperty("column.chunk.alignment"));
+        checkArgument(this.chunkAlignment >= 0, "column.chunk.alignment must >= 0");
+        this.chunkPaddingBuffer = new byte[this.chunkAlignment];
         checkArgument(fileContentLength > 0, "file content length is not positive");
         this.fileContentLength = fileContentLength;
         checkArgument(fileRowNum > 0, "file row number is not positive");
@@ -103,7 +115,7 @@ private PixelsCompactor(
 
         this.fileColStatRecorders = requireNonNull(fileColStatRecorders, "file column stat reader is null");
 
-        checkArgument(!requireNonNull(rowGroupFooterBuilderList).isEmpty());
+        checkArgument(!requireNonNull(rowGroupInfoBuilderList).isEmpty());
         checkArgument(!requireNonNull(rowGroupStatBuilderList).isEmpty());
         checkArgument(!requireNonNull(rowGroupFooterBuilderList).isEmpty());
         checkArgument(!requireNonNull(rowGroupPaths).isEmpty());
@@ -369,10 +381,34 @@ private void writeColumnChunks()
                 fsReader.seek(columnChunkOffset);
                 byte[] chunkBuffer = new byte[(int) columnChunkLength];
                 fsReader.readFully(chunkBuffer);
-                fsWriter.prepare((int) columnChunkLength);
-                long offset = this.fsWriter.append(chunkBuffer, 0, (int) columnChunkLength);
-                columnChunkIndexBuilder.setChunkOffset(offset);
-                // this.fsWriter.flush(); // Issue #192: no need to flush as writing has not finished.
+
+                // Issue #521: prepare for writing the column chunk, and make sure the start offset is aligned.
+                long chunkStartOffset = fsWriter.prepare((int) columnChunkLength);
+                int tryAlign = 0;
+                while (chunkAlignment != 0 && chunkStartOffset % chunkAlignment != 0 && tryAlign++ < 2)
+                {
+                    int alignBytes = (int) (chunkAlignment - chunkStartOffset % chunkAlignment);
+                    this.fsWriter.append(chunkPaddingBuffer, 0, alignBytes);
+                    chunkStartOffset = this.fsWriter.prepare((int) columnChunkLength);
+                }
+                if (tryAlign > 2)
+                {
+                    LOGGER.warn("failed to align the start offset of the column chunk");
+                    throw new IOException("failed to align the start offset of the column chunk");
+                }
+
+                this.fsWriter.append(chunkBuffer, 0, (int) columnChunkLength);
+                /*
+                 * Issue #521:
+                 * It is not necessary pad the column chunk here, as additional bytes are already padded before
+                 * writing this column chunk to ensure chunkStartOffset is aligned. For the last column chunk,
+                 * there is no need to ensure its length is aligned. We only need aligned start offsets.
+                 *
+                 * Also, there is no need to update the column chunk length, pixels reader needs the real length
+                 * of the column chunk.
+                 */
+                columnChunkIndexBuilder.setChunkOffset(chunkStartOffset);
+                // Issue #192: no need to flush fsWriter as writing has not finished.
             }
             catch (IOException e)
             {
@@ -417,8 +453,8 @@ private void writeRowGroupFooters()
 
     private void writeFileTail()
     {
-        PixelsProto.Footer footer = writeFooter();
-        PixelsProto.PostScript postScript = writePostScript();
+        PixelsProto.Footer footer = buildFileFooter();
+        PixelsProto.PostScript postScript = buildPostScript();
 
         PixelsProto.FileTail fileTail =
                 PixelsProto.FileTail.newBuilder()
@@ -447,7 +483,7 @@ private void writeFileTail()
         }
     }
 
-    private PixelsProto.Footer writeFooter()
+    private PixelsProto.Footer buildFileFooter()
     {
         PixelsProto.Footer.Builder footerBuilder =
                 PixelsProto.Footer.newBuilder();
@@ -468,7 +504,7 @@ private PixelsProto.Footer writeFooter()
         return footerBuilder.build();
     }
 
-    private PixelsProto.PostScript writePostScript()
+    private PixelsProto.PostScript buildPostScript()
     {
         return PixelsProto.PostScript.newBuilder()
                 .setVersion(Constants.VERSION)
@@ -478,6 +514,8 @@ private PixelsProto.PostScript writePostScript()
                 .setCompressionBlockSize(compressionBlockSize)
                 .setPixelStride(pixelStride)
                 .setWriterTimezone(timeZone.getDisplayName())
+                .setPartitioned(false) // Issue #521: we do not compact partitioned files.
+                .setColumnChunkAlignment(chunkAlignment)
                 .setMagic(Constants.MAGIC)
                 .build();
     }