[Issue pixelsdb#830] sync the updates from mini-pixels (pixelsdb#832)

Author: bianhq

cpp/Makefile

@@ -46,7 +46,8 @@ deps:
 	make -j install
 
 clean:
-	rm -rf build
+	rm -rf build/release
+	rm -rf build/debug
 	cd pixels-duckdb/duckdb && make clean
 
 # Main build

cpp/pixels-cli/CMakeLists.txt

@@ -76,6 +76,7 @@ set(pixels_cli_cxx
         lib/executor/LoadExecutor.cpp
         lib/load/Parameters.cpp
         lib/load/PixelsConsumer.cpp)
+
 add_executable(pixels-cli ${pixels_cli_cxx})
 include_directories(include)
 include_directories(../pixels-core/include)

cpp/pixels-cli/lib/load/PixelsConsumer.cpp

@@ -61,12 +61,9 @@ void PixelsConsumer::run()
         regex = " ";
     }
 
-//    int pixelsStride = std::stoi(ConfigFactory::Instance().getProperty("pixel.stride"));
-//    int rowGroupSize = std::stoi(ConfigFactory::Instance().getProperty("row.group.size"));
-//    int64_t blockSize = std::stoll(ConfigFactory::Instance().getProperty("block.size"));
-    int pixelsStride = 2;
-    int rowGroupSize = 100;
-    int64_t blockSize = 1024;
+    int pixelsStride = std::stoi(ConfigFactory::Instance().getProperty("pixel.stride"));
+    int rowGroupSize = std::stoi(ConfigFactory::Instance().getProperty("row.group.size"));
+    int64_t blockSize = std::stoll(ConfigFactory::Instance().getProperty("block.size"));
 
     short replication = static_cast<short>(std::stoi(ConfigFactory::Instance().getProperty("block.replication")));
 
@@ -109,16 +106,14 @@ void PixelsConsumer::run()
                 if (initPixelsFile)
                 {
                     LocalFS targetStorage;
-                    targetFileName =
-                            std::to_string(std::chrono::system_clock::to_time_t(std::chrono::system_clock::now())) +
-                            ".pxl";
+                    targetFileName = std::to_string(std::chrono::system_clock::to_time_t(std::chrono::system_clock::now())) + \
+                                         "_" + std::to_string(this->loadedFiles.size()) + ".pxl";
                     targetFilePath = targetPath + targetFileName;
                     pixelsWriter = std::make_shared<PixelsWriterImpl>(schema, pixelsStride, rowGroupSize,
                                                                       targetFilePath, blockSize,
                                                                       true, encodingLevel, nullPadding, false, 1);
                 }
                 initPixelsFile = false;
-
                 ++rowBatch->rowCount;
                 ++rowCounter;
 
@@ -149,7 +144,7 @@ void PixelsConsumer::run()
                     rowBatch->reset();
                 }
 
-                // 创建一个新的文件
+                // create a new file
                 if (rowCounter >= maxRowNum)
                 {
                     if (rowBatch->rowCount != 0)
@@ -159,13 +154,14 @@ void PixelsConsumer::run()
                     }
                     pixelsWriter->close();
                     this->loadedFiles.push_back(targetFilePath);
+                    std::cout << "Generate file: " << targetFilePath << std::endl;
                     rowCounter = 0;
                     initPixelsFile = true;
                 }
             }
         }
     }
-    // 剩余line写入文件
+    // write the remaining lines
     if (rowCounter > 0)
     {
         if (rowBatch->rowCount != 0)

cpp/pixels-common/include/physical/Request.h

@@ -35,7 +35,8 @@ class Request
     uint64_t start;
     uint64_t length;
 
-    Request(uint64_t queryId_, uint64_t start_, uint64_t length_, int64_t bufferId = -1);
+    Request(uint64_t queryId_, uint64_t start_, uint64_t length_,
+            int64_t bufferId = -1);
 
     int hashCode();
 

cpp/pixels-common/include/physical/natives/ByteBuffer.h

@@ -58,11 +58,9 @@ class ByteBuffer
     uint8_t
     get(); // Relative get method. Reads the uint8_t at the buffers current position then increments the position
     uint8_t get(uint32_t index); // Absolute get method. Read uint8_t at index
-    // this is the same as read(byte b[], int off, int len) in InputStream.java
     int getBufferOffset();
-
     uint8_t *getBuffer();
-
+    // this is the same as read(byte b[], int off, int len) in InputStream.java
     int read(uint8_t *buffer, uint32_t off, uint32_t len);
 
     void getBytes(uint8_t *buffer, uint32_t len); // Absolute read into array buf of length len

cpp/pixels-core/CMakeLists.txt

@@ -62,7 +62,11 @@ set(pixels_core_cxx
         lib/writer/TimestampColumnWriter.cpp
         lib/writer/DateColumnWriter.cpp
         include/writer/DateColumnWriter.h
-        )
+        include/utils/DynamicIntArray.h
+        lib/utils/DynamicIntArray.cpp
+        include/writer/StringColumnWriter.h
+        lib/writer/StringColumnWriter.cpp
+)
 
 add_library(pixels-core ${pixels_core_cxx})
 

cpp/pixels-core/include/PixelsWriterImpl.h

@@ -86,6 +86,5 @@ class PixelsWriterImpl : public PixelsWriter
     std::vector <pixels::proto::RowGroupStatistic> rowGroupStatisticList;
     std::shared_ptr <PhysicalWriter> physicalWriter;
     std::vector <std::shared_ptr<TypeDescription>> children;
-
 };
 #endif //PIXELS_PIXELSWRITERIMPL_H

cpp/pixels-core/include/TypeDescription.h

@@ -74,6 +74,7 @@ class TypeDescription : public std::enable_shared_from_this<TypeDescription>
         CHAR,
         STRUCT
     };
+
     class StringPosition
     {
         friend class TypeDescription;

cpp/pixels-core/include/exception/PixelsFileVersionInvalidException.h

@@ -26,6 +26,7 @@
 #define PIXELS_PIXELSRUNTIMEEXCEPTION_H
 
 #include <exception>
+#include <cstdint>
 #include <string>
 #include <iostream>
 #include "PixelsVersion.h"

cpp/pixels-core/include/reader/PixelsRecordReaderImpl.h

@@ -149,5 +149,6 @@ class PixelsRecordReaderImpl : public PixelsRecordReader
     std::shared_ptr <TypeDescription> fileSchema;
     std::shared_ptr <TypeDescription> resultSchema;
     std::shared_ptr <VectorizedRowBatch> resultRowBatch;
+    int asyncReadRequestNum{0};
 };
 #endif //PIXELS_PIXELSRECORDREADERIMPL_H