Feature: PAX support VEC executor

In the vectorized executor, we need to return the batch of rows(for the single column).

In PAX, data needs to be transformed to record batch. This is because the organize of
column data is different with record batch.

- Fixed-length columns: PAX will not pad null datum by length. But record batch requires
  null padding by length.
- Non-fixed-length columns: PAX will store the datum header and use length stream.
  Record batch requires removing the header and using offset array

Author: jiaqizho

contrib/pax_storage/CMakeLists.txt

@@ -56,4 +56,17 @@ else()
   SET(BUILD_GTEST OFF)
 endif(ENBALE_DEBUG)
 
+# Vec options
+option(VEC_BUILD "Build pax vectorization version" OFF)
+set(VEC_HOME "" CACHE STRING "Path to vectorization home")
+if (VEC_BUILD)
+
+if("${VEC_HOME}" STREQUAL "")
+  message(FATAL_ERROR "No found vectorization home setting. Using -DVEC_HOME to spec vectorization home")
+endif()
+
+set(CBDB_ROOT_INCLUDE_DIR ${PG_INCLUDEDIR})
+ADD_DEFINITIONS(-DVEC_BUILD)
+
+endif(VEC_BUILD)
 add_subdirectory(src/cpp)

contrib/pax_storage/src/cpp/CMakeLists.txt

@@ -173,6 +173,9 @@ set(pax_catalog_src
     catalog/micro_partition_stats.cc
     catalog/pax_aux_table.cc)
 
+set(pax_vec_src
+  storage/vec/pax_vec_adapter.cc)
+
 link_directories($ENV{GPHOME}/lib)
 
 if(BUILD_PAX_FORMAT)
@@ -185,14 +188,11 @@ set_target_properties(paxformat PROPERTIES
 OUTPUT_NAME paxformat)
 add_dependencies(paxformat generate_protobuf)
 
-
 # export headers
 set(PAX_COMM_HEADERS
   comm/cbdb_api.h
 )
 
-
-
 ## install dynamic libraray 
 install(TARGETS paxformat
   LIBRARY DESTINATION ${CMAKE_INSTALL_PREFIX}/lib)
@@ -212,10 +212,10 @@ install(FILES ${PAX_COMM_HEADERS}
 )
 
 else()
-add_library(pax SHARED ${orc_proto_src} ${pax_proto_src} ${pax_storage_src} ${stats_proto_src} ${pax_exceptions_src} ${pax_access_src} ${pax_comm_src} ${pax_catalog_src})
+add_library(pax SHARED ${orc_proto_src} ${pax_proto_src} ${pax_storage_src} ${stats_proto_src} ${pax_exceptions_src} ${pax_access_src} ${pax_comm_src} ${pax_catalog_src} ${pax_vec_src})
 set_target_properties(pax PROPERTIES 
 OUTPUT_NAME pax)
-target_include_directories(pax PUBLIC ${ZTSD_HEADER}  ${CMAKE_CURRENT_SOURCE_DIR} "${CBDB_INCLUDE_DIR}")
+target_include_directories(pax PUBLIC ${ZTSD_HEADER}  ${CMAKE_CURRENT_SOURCE_DIR} ${CBDB_INCLUDE_DIR})
 target_link_libraries(pax PUBLIC uuid orc_protobuf zstd z postgres)
 add_dependencies(pax generate_protobuf)
 add_dependencies(pax stats_generate_protobuf)
@@ -224,3 +224,20 @@ add_custom_command(TARGET pax POST_BUILD
                        copy_if_different $<TARGET_FILE:pax> ${CMAKE_CURRENT_SOURCE_DIR}/../data/pax.so)
 endif(BUILD_PAX_FORMAT)
 
+# vec build
+if (VEC_BUILD)
+  set(VEC_HEADER ${VEC_HOME}/src/include/)
+
+  find_package(PkgConfig REQUIRED)
+  pkg_check_modules(GLIB REQUIRED glib-2.0)
+
+  target_include_directories(pax PRIVATE 
+    ${VEC_HEADER} # for utils/tuptable_vec.h
+    ${CBDB_ROOT_INCLUDE_DIR}  # for arrow-glib/arrow-glib.h and otehr arrow interface
+    ${GLIB_INCLUDE_DIRS} # for glib-object.h
+  )
+if(BUILD_GTEST)
+  target_include_directories(test_main PRIVATE ${VEC_HEADER} ${CBDB_ROOT_INCLUDE_DIR} ${GLIB_INCLUDE_DIRS})
+endif(BUILD_GTEST)
+  target_link_libraries(pax PRIVATE arrow)
+endif(VEC_BUILD)

contrib/pax_storage/src/cpp/access/pax_scanner.cc

@@ -67,6 +67,12 @@ TableScanDesc PaxScanDesc::BeginScan(Relation relation, Snapshot snapshot,
       new TableReader(micro_partition_reader, std::move(iter), reader_options);
   desc->reader_->Open();
 
+#ifdef VEC_BUILD
+  if (flags & (1 << 12)) {
+    desc->vec_adapter_ = new VecAdapter(cbdb::RelationGetTupleDesc(relation));
+  }
+#endif
+
   MemoryContextSwitchTo(old_ctx);
   return &desc->rs_base_;
 }
@@ -81,6 +87,9 @@ void PaxScanDesc::EndScan(TableScanDesc scan) {
   delete desc->reader_;
   if (desc->filter_) delete desc->filter_;
 
+#ifdef VEC_BUILD
+  delete desc->vec_adapter_;
+#endif
   // TODO(jiaqizho): please double check with abort transaction @gongxun
   Assert(desc->memory_context_);
   cbdb::MemoryCtxDelete(desc->memory_context_);
@@ -144,7 +153,18 @@ bool PaxScanDesc::ScanGetNextSlot(TableScanDesc scan, TupleTableSlot *slot) {
 
   CTupleSlot cslot(slot);
   old_ctx = MemoryContextSwitchTo(desc->memory_context_);
+
+#ifdef VEC_BUILD
+  // TODO(vec): replace (1 << 12), and check other place
+  if (desc->rs_base_.rs_flags & (1 << 12)) {
+    ok = desc->reader_->ReadVecTuple(&cslot, desc->vec_adapter_);
+    MemoryContextSwitchTo(old_ctx);
+    return ok;
+  }
+#endif
+
   ok = desc->reader_->ReadTuple(&cslot);
+
   MemoryContextSwitchTo(old_ctx);
   return ok;
 }

contrib/pax_storage/src/cpp/access/pax_scanner.h

@@ -4,7 +4,9 @@
 
 #include "storage/pax.h"
 #include "storage/pax_filter.h"
-
+#ifdef VEC_BUILD
+#include "storage/vec/pax_vec_adapter.h"
+#endif
 namespace pax {
 
 class PaxScanDesc {
@@ -62,6 +64,9 @@ class PaxScanDesc {
 
   // filter used to do column projection
   PaxFilter *filter_ = nullptr;
+#ifdef VEC_BUILD
+  VecAdapter *vec_adapter_ = nullptr;
+#endif
 };  // class PaxScanDesc
 
 }  // namespace pax

contrib/pax_storage/src/cpp/comm/cbdb_wrappers.cc

@@ -317,6 +317,12 @@ int cbdb::RelationGetAttributesNumber(Relation rel) {
   CBDB_WRAP_END;
 }
 
+TupleDesc cbdb::RelationGetTupleDesc(Relation rel) {
+  CBDB_WRAP_START;
+  { return RelationGetDescr(rel); }
+  CBDB_WRAP_END;
+}
+
 bool cbdb::ExtractcolumnsFromNode(Node *expr, bool *cols, AttrNumber natts) {
   CBDB_WRAP_START;
   { return extractcolumns_from_node(expr, cols, natts); }

contrib/pax_storage/src/cpp/comm/cbdb_wrappers.h

@@ -139,6 +139,8 @@ std::string BuildPaxDirectoryPath(RelFileNode rd_node, BackendId rd_backend);
 
 int RelationGetAttributesNumber(Relation rel);
 
+TupleDesc RelationGetTupleDesc(Relation rel);
+
 bool ExtractcolumnsFromNode(Node *expr, bool *cols, AttrNumber natts);
 
 std::string BuildPaxFilePath(Relation rel, const std::string &block_id);

contrib/pax_storage/src/cpp/storage/columns/pax_column.cc

@@ -333,7 +333,7 @@ std::pair<char *, size_t> PaxNonFixedColumn::GetBuffer(size_t position) {
 
 std::pair<char *, size_t> PaxNonFixedColumn::GetRangeBuffer(size_t start_pos,
                                                             size_t len) {
-  CBDB_CHECK((start_pos + len) <= GetNonNullRows() && len > 0,
+  CBDB_CHECK((start_pos + len) <= GetNonNullRows() && len >= 0,
              cbdb::CException::ExType::kExTypeOutOfRange);
   size_t range_len = 0;
 

contrib/pax_storage/src/cpp/storage/pax.cc

@@ -127,8 +127,7 @@ TableReader::TableReader(
       is_empty_(true),
       reader_options_(options),
       table_no_(0),
-      table_index_(0) {
-}
+      table_index_(0) {}
 
 TableReader::~TableReader() {
   if (reader_) {
@@ -163,8 +162,11 @@ void TableReader::Close() {
   if (is_empty_) {
     return;
   }
-  Assert(reader_);
-  reader_->Close();
+
+  if (reader_) {
+    reader_->Close();
+    reader_ = nullptr;
+  }
 }
 
 bool TableReader::ReadTuple(CTupleSlot *slot) {
@@ -188,6 +190,59 @@ bool TableReader::ReadTuple(CTupleSlot *slot) {
   return true;
 }
 
+#ifdef VEC_BUILD
+// TODO(jiaqizho): should remove this method but provider a
+// new mirco partition reader which include adapter
+bool TableReader::ReadVecTuple(CTupleSlot *slot, VecAdapter *adapter) {
+  size_t flush_nums_of_rows = 0;
+
+  if (is_empty_) {
+    return false;
+  }
+
+  // Three conditions indicate that there is no data left
+  // - `VecAdapter` must be set
+  // - `VecAdapter` has consumed all buffer data
+  // - current have not next reader
+  if (adapter->IsInitialized() && adapter->IsEnd() && !iterator_->HasNext()) {
+    return false;
+  }
+
+  Assert(reader_);
+
+  // `No set` means that the first reader not setup in apapter
+  if (!adapter->IsInitialized()) {
+    PaxColumns *pax_columns =
+        reinterpret_cast<OrcIteratorReader *>(reader_)->GetAllColumns();
+    adapter->SetDataSource(pax_columns);
+  }
+
+  // `VecAdapter` has consumed all buffer data
+  if (adapter->IsEnd()) {
+    if (iterator_->HasNext()) {
+      reader_->Close();
+      OpenFile();
+
+      PaxColumns *pax_columns =
+          reinterpret_cast<OrcIteratorReader *>(reader_)->GetAllColumns();
+      adapter->SetDataSource(pax_columns);
+    }
+  }
+
+  auto ok = adapter->AppendToVecBuffer();
+  CBDB_CHECK(ok, cbdb::CException::ExType::kExTypeLogicError);
+
+  flush_nums_of_rows = adapter->FlushVecBuffer(slot);
+  Assert(flush_nums_of_rows);
+
+  num_tuples_ += flush_nums_of_rows;
+  slot->SetTableNo(table_no_);
+  slot->SetBlockNumber(current_block_number_);
+  slot->StoreVirtualTuple();
+  return true;
+}
+#endif  // VEC_BUILD
+
 void TableReader::OpenFile() {
   Assert(iterator_->HasNext());
   auto it = iterator_->Next();

contrib/pax_storage/src/cpp/storage/pax.h

@@ -19,6 +19,10 @@
 #include "storage/paxc_block_map_manager.h"
 #include "storage/strategy.h"
 
+#ifdef VEC_BUILD
+#include "storage/vec/pax_vec_adapter.h"
+#endif
+
 namespace pax {
 
 class TableWriter {
@@ -80,6 +84,11 @@ class TableReader final {
 
   void Close();
 
+#ifdef VEC_BUILD
+  // std::unique_ptr<VecAdapter> here ?
+  bool ReadVecTuple(CTupleSlot *slot, VecAdapter *adapter);
+#endif
+
   bool ReadTuple(CTupleSlot *slot);
 
   // deprecate:

contrib/pax_storage/src/cpp/storage/vec/arrow_wrapper.h

@@ -0,0 +1,58 @@
+#pragma once
+
+#ifdef VEC_BUILD
+
+// FIXME(jiaqizho): There marco defined in datatime.h
+// which include in `cbdb_api.h`. In pax, we always need
+// include `cbdb_api.h`.
+
+#undef RESERV
+#undef MONTH
+#undef YEAR
+#undef DAY
+#undef JULIAN
+#undef TZ
+#undef DTZ
+#undef DYNTZ
+#undef IGNORE_DTF
+#undef AMPM
+#undef HOUR
+#undef MINUTE
+#undef SECOND
+#undef MILLISECOND
+#undef MICROSECOND
+#undef IsPowerOf2
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wunused-variable"
+#pragma GCC diagnostic ignored "-Wunused-but-set-variable"
+
+#include <arrow/array/array_binary.h>
+#include <arrow/array/array_nested.h>
+#include <arrow/array/array_primitive.h>
+#include <arrow/c/abi.h>
+#include <arrow/c/bridge.h>
+#include <arrow/util/bit_util.h>
+
+#pragma GCC diagnostic pop
+
+#define RESERV 0
+#define MONTH 1
+#define YEAR 2
+#define DAY 3
+#define JULIAN 4
+#define TZ 5
+#define DTZ 6
+#define DYNTZ 7
+#define IGNORE_DTF 8
+#define AMPM 9
+#define HOUR 10
+#define MINUTE 11
+#define SECOND 12
+#define MILLISECOND 13
+#define MICROSECOND 14
+
+// NOLINTNEXTLINE
+#define IsPowerOf2(x) (x > 0 && ((x) & ((x)-1)) == 0)
+
+#endif  // VEC_BUILD