ADD BE UDF bitmap_and and bitmap_or

be/src/exec/parquet_reader.cpp

@@ -49,8 +49,13 @@ ParquetReaderWrap::~ParquetReaderWrap() {
 Status ParquetReaderWrap::init_parquet_reader(const std::vector<SlotDescriptor*>& tuple_slot_descs) {
     try {
         // new file reader for parquet file
-        _reader.reset(new parquet::arrow::FileReader(arrow::default_memory_pool(),
-                std::move(parquet::ParquetFileReader::Open(_parquet, _properties))));
+        auto st = parquet::arrow::FileReader::Make(arrow::default_memory_pool(),
+                                                   parquet::ParquetFileReader::Open(_parquet, _properties),
+                                                   &_reader);
+        if (!st.ok()) {
+            LOG(WARNING) << "failed to create parquet file reader, errmsg=" << st.ToString();
+            return Status::InternalError("Failed to create file reader");
+        }
 
         _file_metadata = _reader->parquet_reader()->metadata();
         // initial members
@@ -511,6 +516,7 @@ arrow::Status ParquetFile::ReadAt(int64_t position, int64_t nbytes, int64_t* byt
         position += reads;
         out = (char*)out + reads;
     }
+    _pos += *bytes_read;
     return arrow::Status::OK();
 }
 
@@ -531,7 +537,17 @@ arrow::Status ParquetFile::Tell(int64_t* position) const {
 }
 
 arrow::Status ParquetFile::Read(int64_t nbytes, std::shared_ptr<arrow::Buffer>* out) {
-    return arrow::Status::NotImplemented("Not Supported.");
+    std::shared_ptr<arrow::Buffer> read_buf;
+    ARROW_RETURN_NOT_OK(arrow::AllocateBuffer(arrow::default_memory_pool(), nbytes, &read_buf));
+    int64_t bytes_read = 0;
+    ARROW_RETURN_NOT_OK(ReadAt(_pos, nbytes, &bytes_read, read_buf->mutable_data()));
+    // If bytes_read is equal with read_buf's capacity, we just assign
+    if (bytes_read == nbytes) {
+        *out = std::move(read_buf);
+    } else {
+        *out = arrow::SliceBuffer(read_buf, 0, bytes_read);
+    }
+    return arrow::Status::OK();
 }
 
 }

be/src/exec/parquet_reader.h

@@ -63,6 +63,7 @@ class ParquetFile : public arrow::io::RandomAccessFile {
     bool closed() const override;
 private:
     FileReader *_file;
+    int64_t _pos = 0;
 };
 
 // Reader of broker parquet file

be/src/exprs/bitmap_function.cpp

@@ -388,6 +388,51 @@ BigIntVal BitmapFunctions::bitmap_intersect_finalize(FunctionContext* ctx, const
     return result;
 }
 
+StringVal BitmapFunctions::bitmap_or(FunctionContext* ctx, const StringVal& src, const StringVal& dst){
+    RoaringBitmap bitmap;
+    if(!src.is_null){
+        if(src.len == 0 ){
+            bitmap.merge(*reinterpret_cast<RoaringBitmap*>(src.ptr));
+        } else{
+            bitmap.merge(RoaringBitmap ((char*)src.ptr));
+        }
+    }
+
+    if(!dst.is_null){
+        if(dst.len == 0){
+            bitmap.merge(*reinterpret_cast<RoaringBitmap*>(dst.ptr));
+        } else{
+            bitmap.merge(RoaringBitmap ((char*)dst.ptr));
+        }
+    }
+
+    StringVal result(ctx,bitmap.size());
+    bitmap.serialize((char*)result.ptr);
+    return result;
+}
+StringVal BitmapFunctions::bitmap_and(FunctionContext* ctx, const StringVal& src, const StringVal& dst){
+    RoaringBitmap bitmap;
+    if(!src.is_null){
+        if(src.len == 0 ){
+            bitmap.merge(*reinterpret_cast<RoaringBitmap*>(src.ptr));
+        } else{
+            bitmap.merge(RoaringBitmap ((char*)src.ptr));
+        }
+    }
+
+    if(!dst.is_null){
+        if(dst.len == 0){
+            bitmap.intersect(*reinterpret_cast<RoaringBitmap*>(dst.ptr));
+        } else{
+            bitmap.intersect(RoaringBitmap ((char*)dst.ptr));
+        }
+    }
+
+    StringVal result(ctx,bitmap.size());
+    bitmap.serialize((char*)result.ptr);
+    return result;
+}
+
 
 template void BitmapFunctions::bitmap_update_int<TinyIntVal>(
         FunctionContext* ctx, const TinyIntVal& src, StringVal* dst);

be/src/exprs/bitmap_function.h

@@ -41,6 +41,8 @@ class BitmapFunctions {
     static StringVal to_bitmap(FunctionContext* ctx, const StringVal& src);
     static StringVal bitmap_hash(FunctionContext* ctx, const StringVal& src);
 
+    static StringVal bitmap_or(FunctionContext* ctx, const StringVal& src,const StringVal& dst);
+    static StringVal bitmap_and(FunctionContext* ctx, const StringVal& src,const StringVal& dst);
 
     // bitmap_intersect
     template<typename T, typename ValType>

be/src/exprs/hll_function.cpp

@@ -80,6 +80,12 @@ BigIntVal HllFunctions::hll_finalize(FunctionContext*, const StringVal &src) {
     return result;
 }
 
+BigIntVal HllFunctions::hll_get_value(FunctionContext*, const StringVal &src) {
+    auto* src_hll = reinterpret_cast<HyperLogLog*>(src.ptr);
+    BigIntVal result(src_hll->estimate_cardinality());
+    return result;
+}
+
 BigIntVal HllFunctions::hll_cardinality(FunctionContext* ctx, const StringVal& input) {
     if (input.is_null) {
         return BigIntVal::null();

be/src/exprs/hll_function.h

@@ -36,6 +36,10 @@ class HllFunctions {
 
     static BigIntVal hll_finalize(FunctionContext*, const StringVal& src);
 
+    // Get the hll cardinality, the difference from hll_finalize method is
+    // hll_get_value method doesn't free memory
+    static BigIntVal hll_get_value(FunctionContext*, const StringVal& src);
+
     static StringVal hll_serialize(FunctionContext* ctx, const StringVal& src);
 
     static BigIntVal hll_cardinality(FunctionContext* ctx, const StringVal& src);

be/src/olap/olap_snapshot_converter.cpp

@@ -335,6 +335,9 @@ OLAPStatus OlapSnapshotConverter::to_column_pb(const ColumnMessage& column_msg,
     if (column_msg.has_is_bf_column()) {
         column_pb->set_is_bf_column(column_msg.is_bf_column());
     }
+    if (column_msg.has_has_bitmap_index()) {
+        column_pb->set_has_bitmap_index(column_msg.has_bitmap_index());
+    }
     // TODO(ygl) calculate column id from column list
     // column_pb->set_referenced_column_id(column_msg.());
 
@@ -388,6 +391,9 @@ OLAPStatus OlapSnapshotConverter::to_column_msg(const ColumnPB& column_pb, Colum
     if (column_pb.has_is_bf_column()) {
         column_msg->set_is_bf_column(column_pb.is_bf_column());
     }
+    if (column_pb.has_has_bitmap_index()) {
+        column_msg->set_has_bitmap_index(column_pb.has_bitmap_index());
+    }
     column_msg->set_is_root_column(true);
     return OLAP_SUCCESS;
 }

be/src/olap/rowset/segment_v2/segment_iterator.cpp

@@ -520,6 +520,7 @@ Status SegmentIterator::next_batch(RowBlockV2* block) {
             column_predicate->evaluate(&column_block, block->selection_vector(), &selected_size);
         }
         block->set_selected_size(selected_size);
+        block->set_num_rows(selected_size);
         _opts.stats->rows_vec_cond_filtered += original_size - selected_size;
     }
 

be/src/olap/rowset/segment_v2/segment_writer.cpp

@@ -62,10 +62,6 @@ Status SegmentWriter::init(uint32_t write_mbytes_per_sec) {
         if (column.is_key()) {
             opts.need_zone_map = true;
         }
-        // TODO set opts.need_bitmap_index based on table properties
-        if (!column.is_key()) {
-            opts.need_bitmap_index = _opts.need_bitmap_index;
-        }
         if (column.is_bf_column()) {
             opts.need_bloom_filter = true;
             if ((column.aggregation() == OLAP_FIELD_AGGREGATION_REPLACE
@@ -77,6 +73,17 @@ Status SegmentWriter::init(uint32_t write_mbytes_per_sec) {
                 opts.need_bloom_filter = false;
             }
         }
+        if (column.has_bitmap_index()) {
+            opts.need_bitmap_index = true;
+            if ((column.aggregation() == OLAP_FIELD_AGGREGATION_REPLACE
+                 || column.aggregation() == OLAP_FIELD_AGGREGATION_REPLACE_IF_NOT_NULL)
+                && !_opts.whether_to_filter_value) {
+                // if the column's Aggregation type is OLAP_FIELD_AGGREGATION_REPLACE or
+                // OLAP_FIELD_AGGREGATION_REPLACE_IF_NOT_NULL and the segment is not in base rowset,
+                // do not write the bitmap index because it is useless
+                opts.need_bitmap_index = false;
+            }
+        }
 
         std::unique_ptr<Field> field(FieldFactory::create(column));
         DCHECK(field.get() != nullptr);

be/src/olap/rowset/segment_v2/segment_writer.h

@@ -44,8 +44,6 @@ extern const uint32_t k_segment_magic_length;
 
 struct SegmentWriterOptions {
     uint32_t num_rows_per_block = 1024;
-    // Todo(kks): only for UT, we should remove it when we support bitmap_index in FE
-    bool need_bitmap_index = false;
     // whether to filter value column against bloom filter/zone map
     bool whether_to_filter_value = false;
 };

be/src/olap/schema_change.cpp

@@ -1931,6 +1931,10 @@ OLAPStatus SchemaChangeHandler::_parse_request(TabletSharedPtr base_tablet,
                        != ref_tablet_schema.column(column_mapping->ref_column).is_bf_column()) {
                 *sc_directly = true;
                 return OLAP_SUCCESS;
+            }  else if (new_tablet_schema.column(i).has_bitmap_index()
+                        != ref_tablet_schema.column(column_mapping->ref_column).has_bitmap_index()) {
+                *sc_directly = true;
+                return OLAP_SUCCESS;
             }
         }
     }

be/src/olap/tablet_manager.cpp

@@ -1198,7 +1198,17 @@ OLAPStatus TabletManager::_create_inital_rowset_unlocked(
             context.tablet_id = tablet->tablet_id();
             context.partition_id = tablet->partition_id();
             context.tablet_schema_hash = tablet->schema_hash();
-            context.rowset_type = StorageEngine::instance()->default_rowset_type();
+            if (!request.__isset.storage_format || request.storage_format == TStorageFormat::DEFAULT) {
+                context.rowset_type = StorageEngine::instance()->default_rowset_type();
+            } else if (request.storage_format == TStorageFormat::V1){
+                context.rowset_type = RowsetTypePB::ALPHA_ROWSET;
+            } else if (request.storage_format == TStorageFormat::V2) {
+                context.rowset_type = RowsetTypePB::BETA_ROWSET;
+            } else {
+                LOG(ERROR) << "invalid TStorageFormat: " << request.storage_format;
+                DCHECK(false);
+                context.rowset_type = StorageEngine::instance()->default_rowset_type();
+            }
             context.rowset_path_prefix = tablet->tablet_path();
             context.tablet_schema = &(tablet->tablet_schema());
             context.rowset_state = VISIBLE;

be/src/olap/tablet_meta.cpp

@@ -18,6 +18,7 @@
 #include "olap/tablet_meta.h"
 
 #include <sstream>
+#include <boost/algorithm/string.hpp>
 
 #include "olap/file_helper.h"
 #include "olap/olap_common.h"
@@ -116,6 +117,7 @@ TabletMeta::TabletMeta(int64_t table_id, int64_t partition_id,
         uint32_t unique_id = col_ordinal_to_unique_id.at(col_ordinal++);
         column->set_unique_id(unique_id);
         column->set_name(tcolumn.column_name);
+        column->set_has_bitmap_index(false);
         string data_type;
         EnumToString(TPrimitiveType, tcolumn.column_type.type, data_type);
         column->set_type(data_type);
@@ -152,6 +154,17 @@ TabletMeta::TabletMeta(int64_t table_id, int64_t partition_id,
             column->set_is_bf_column(tcolumn.is_bloom_filter_column);
             has_bf_columns = true;
         }
+        if (tablet_schema.__isset.indexes) {
+            for (auto& index : tablet_schema.indexes) {
+                if (index.index_type == TIndexType::type::BITMAP) {
+                    DCHECK_EQ(index.columns.size(), 1);
+                    if (boost::iequals(tcolumn.column_name, index.columns[0])) {
+                        column->set_has_bitmap_index(true);
+                        break;
+                    }
+                }
+            }
+        }
     }
 
     schema->set_next_column_unique_id(next_unique_id);

be/src/olap/tablet_schema.cpp

@@ -295,6 +295,11 @@ OLAPStatus TabletColumn::init_from_pb(const ColumnPB& column) {
     } else {
         _is_bf_column = false;
     }
+    if (column.has_has_bitmap_index()) {
+        _has_bitmap_index = column.has_bitmap_index();
+    } else {
+        _has_bitmap_index = false;
+    }
     _has_referenced_column = column.has_referenced_column_id();
     if (_has_referenced_column) {
         _referenced_column_id = column.referenced_column_id();
@@ -327,6 +332,9 @@ OLAPStatus TabletColumn::to_schema_pb(ColumnPB* column) {
     if (_has_referenced_column) {
         column->set_referenced_column_id(_referenced_column_id);
     }
+    if (_has_bitmap_index) {
+        column->set_has_bitmap_index(_has_bitmap_index);
+    }
     return OLAP_SUCCESS;
 }
 

be/src/olap/tablet_schema.h

@@ -40,6 +40,7 @@ class TabletColumn {
     inline bool is_key() const { return _is_key; }
     inline bool is_nullable() const { return _is_nullable; }
     inline bool is_bf_column() const { return _is_bf_column; }
+    inline bool has_bitmap_index() const {return _has_bitmap_index; }
     bool has_default_value() const { return _has_default_value; }
     std::string default_value() const { return _default_value; }
     bool has_reference_column() const { return _has_referenced_column; }
@@ -80,6 +81,8 @@ class TabletColumn {
     bool _has_referenced_column;
     int32_t _referenced_column_id;
     std::string _referenced_column;
+
+    bool _has_bitmap_index = false;
 };
 
 class TabletSchema {

be/test/exprs/bitmap_function_test.cpp

@@ -28,19 +28,17 @@
 
 namespace doris {
 
-StringVal convert_bitmap_to_string(FunctionContext* ctx, RoaringBitmap& bitmap) {
-    std::string buf;
-    buf.resize(bitmap.size());
-    bitmap.serialize((char*)buf.c_str());
-    return AnyValUtil::from_string_temp(ctx, buf);
+StringVal convert_bitmap_to_string(FunctionContext* ctx,RoaringBitmap& bitmap) {
+    StringVal result(ctx, bitmap.size());
+    bitmap.serialize((char*)result.ptr);
+    return result;
 }
 
 template<typename T>
 StringVal convert_bitmap_intersect_to_string(FunctionContext* ctx, BitmapIntersect<T>& intersect) {
-    std::string buf;
-    buf.resize(intersect.size());
-    intersect.serialize((char*)buf.c_str());
-    return AnyValUtil::from_string_temp(ctx, buf);
+    StringVal result(ctx,intersect.size());
+    intersect.serialize((char*)result.ptr);
+    return result;
 }
 
 class BitmapFunctionsTest : public testing::Test {
@@ -248,6 +246,46 @@ TEST_F(BitmapFunctionsTest, test_bitmap_intersect) {
 }
 
 
+TEST_F(BitmapFunctionsTest,bitmap_or) {
+    RoaringBitmap bitmap1(1024);
+    bitmap1.update(1);
+    bitmap1.update(2019);
+
+    RoaringBitmap bitmap2(33);
+    bitmap2.update(44);
+    bitmap2.update(55);
+
+    StringVal bitmap_src = convert_bitmap_to_string(ctx, bitmap1);
+    StringVal bitmap_dst = convert_bitmap_to_string(ctx, bitmap2);
+
+    StringVal bitmap_str = BitmapFunctions::bitmap_or(ctx,bitmap_src,bitmap_dst);
+    BigIntVal result = BitmapFunctions::bitmap_count(ctx,bitmap_str);
+
+    BigIntVal expected(6);
+    ASSERT_EQ(expected, result);
+}
+
+
+TEST_F(BitmapFunctionsTest,bitmap_and) {
+    RoaringBitmap bitmap1(1024);
+    bitmap1.update(1);
+    bitmap1.update(2019);
+
+    RoaringBitmap bitmap2(33);
+    bitmap2.update(44);
+    bitmap2.update(2019);
+
+    StringVal bitmap_src = convert_bitmap_to_string(ctx, bitmap1);
+    StringVal bitmap_dst = convert_bitmap_to_string(ctx, bitmap2);
+
+    StringVal bitmap_str = BitmapFunctions::bitmap_and(ctx,bitmap_src,bitmap_dst);
+    BigIntVal result = BitmapFunctions::bitmap_count(ctx,bitmap_str);
+
+    BigIntVal expected(1);
+    ASSERT_EQ(expected, result);
+}
+
+
 }
 
 int main(int argc, char** argv) {