modifying ci_scripts/install_pmem_common.sh oap-ape/README.md

ci_scripts/install_pmem_common.sh

@@ -1,4 +1,5 @@
 cd /tmp
 git clone https://github.com/oap-project/pmem-common.git
 cd pmem-common/
+git checkout branch-1.1-spark-3.x
 mvn install -am -q -DskipTests

oap-ape/README.md

@@ -72,7 +72,7 @@ mvn clean install
 Native and Java build. We will use `$OAP_ROOT_DIR/oap-ape/ape-java/ape-spark/target/ape-spark-1.1.0-SNAPSHOT.jar`, `$OAP_ROOT_DIR/oap-ape/ape-java/ape-flink/ape-flink-x.xx.x/target/ape-flink-x.xx.x-1.1.0-SNAPSHOT.jar` later.
 ```
 cd $OAP_ROOT_DIR/oap-ape/ape-java
-mvn clean package -am -Pshading
+mvn clean package -am --batch-mode -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn
 ```
 
 Note: This will package libparquet_jni.so into ape-spark-1.1.0-SNAPSHOT.jar under linux/64/lib folder.

oap-ape/ape-native/src/test/predicateFilterTest.cc

@@ -0,0 +1,170 @@
+#include <stdlib.h>
+
+#include <iostream>
+#include <memory>
+
+#include <arrow/api.h>
+#include <arrow/filesystem/api.h>
+#include <parquet/api/reader.h>
+#include <gtest/gtest.h>
+#include <vector>
+
+#include "src/reader.h"
+#include "src/utils/PredicateExpression.h"
+
+TEST(predicateFilterTest, minMaxTest)
+{
+    arrow::fs::HdfsOptions options_;
+
+    std::string hdfs_host = "clx06-AEP";
+    int hdfs_port = 8020;
+
+    options_.ConfigureEndPoint(hdfs_host, hdfs_port);
+
+    auto result = arrow::fs::HadoopFileSystem::Make(options_);
+    EXPECT_TRUE(result.ok()) << "HadoopFileSystem Make failed";
+
+    std::shared_ptr<arrow::fs::FileSystem> fs_ =
+        std::make_shared<arrow::fs::SubTreeFileSystem>("", *result);
+
+    std::string file_name =
+        "/user/hive/warehouse/tpcds_hdfs_parquet_10.db/store_sales/ss_sold_date_sk=2450817/"
+        "part-00013-0828d1ab-ef1f-4b55-bf94-c071fb76c353.c000.snappy.parquet";
+
+    auto file_result = fs_->OpenInputFile(file_name);
+    EXPECT_TRUE(file_result.ok()) << "Open hdfs file failed";
+
+    std::shared_ptr<arrow::io::RandomAccessFile> file = file_result.ValueOrDie();
+    std::cout << "file size is " << file->GetSize().ValueOrDie() << std::endl;
+
+    parquet::ReaderProperties properties;
+    // std::shared_ptr<parquet::FileMetaData> metadata;
+    std::unique_ptr<parquet::ParquetFileReader> parquetReader =
+        parquet::ParquetFileReader::Open(file, properties, NULLPTR);
+
+    std::shared_ptr<parquet::FileMetaData> fileMetaData = parquetReader->metadata();
+
+    int numRows = fileMetaData->num_rows();
+    int numCols = fileMetaData->num_columns();
+    int numRowGroups = fileMetaData->num_row_groups();
+    std::string schema = fileMetaData->schema()->ToString();
+
+    std::cout<<"parquet file has "<<numRows<<" rows"<<std::endl
+                                <<numCols<<" cols"<<std::endl
+                                <<numRowGroups<<" row groups"<<std::endl;
+
+    std::cout<<"schema: "<<schema<<std::endl;
+
+    if(numRowGroups < 1)
+    {
+        std::cout<<"not enough row groups for test"<<std::endl;
+        return;
+    }
+
+    int rowGroupIndex = 0;
+    std::unique_ptr<parquet::RowGroupMetaData> urgMataData = fileMetaData->RowGroup(rowGroupIndex);
+    std::shared_ptr<parquet::RowGroupMetaData> rgMataData = std::move(urgMataData);
+
+    //std::vector<std::unique_ptr<parquet::ColumnChunkMetaData>> columnChunkMeta;
+    //columnChunkMeta.resize(numCols);
+    std::unique_ptr<parquet::ColumnChunkMetaData> columnChunkMeta;
+
+    for(int i=0; i<numCols; i++)
+    {
+        columnChunkMeta = rgMataData->ColumnChunk(i);
+        std::string column_name = fileMetaData->schema()->Column(i)->name();
+        parquet::Type::type column_type = fileMetaData->schema()->Column(i)->physical_type();
+        std::cout<<"current column["<<i<<"]: "<<column_name
+                                    <<" type "<<column_type
+                                    <<std::endl;
+
+        std::shared_ptr<parquet::Statistics> statistic = columnChunkMeta->statistics();
+        if(!statistic->HasMinMax())
+        {
+            std::cout<<"This column does not have valid min max value."<<std::endl;
+            continue;
+        }
+        switch (column_type)
+        {
+        case parquet::Type::BOOLEAN:{
+            auto boolStatistic = std::static_pointer_cast<parquet::BoolStatistics>(statistic);
+            std::cout<<"min: "<<boolStatistic->min()<<" max: "<<boolStatistic->max()<<std::endl;
+            break;}
+        case parquet::Type::INT32:{
+            auto int32Statistic = std::static_pointer_cast<parquet::Int32Statistics>(statistic);
+            std::cout<<"min: "<<int32Statistic->min()<<" max: "<<int32Statistic->max()<<std::endl;
+            break;}
+        case parquet::Type::INT64:{
+            auto int64Statistic = std::static_pointer_cast<parquet::Int64Statistics>(statistic);
+            std::cout<<"min: "<<int64Statistic->min()<<" max: "<<int64Statistic->max()<<std::endl;
+            break;}
+        case parquet::Type::FLOAT:{
+            auto floatStatistic = std::static_pointer_cast<parquet::FloatStatistics>(statistic);
+            std::cout<<"min: "<<floatStatistic->min()<<" max: "<<floatStatistic->max()<<std::endl;
+            break;}
+
+        default:
+            break;
+        }
+
+    }
+
+}
+
+TEST(predicateFilterTest, predicateTest) {
+
+    std::string hdfs_host = "clx06-AEP";
+    int hdfs_port = 8020;
+
+    std::string file_name =
+        "/user/hive/warehouse/tpcds_hdfs_parquet_10.db/store_sales/ss_sold_date_sk=2450817/"
+        "part-00013-0828d1ab-ef1f-4b55-bf94-c071fb76c353.c000.snappy.parquet";
+
+    std::string requiredColumnName = R"(
+      {
+          "name":"my_schema",
+          "fields":[
+              {
+                  "name":"ss_sold_time_sk",
+                  "value":true
+              },
+              {
+                  "name":"ss_sold_time_sk",
+                  "value":true
+              }
+          ]
+      }
+    )";
+
+    std::string filterJson = R"(
+        {
+            "FilterTypeName":"and",
+            "LeftNode":{
+            "FilterTypeName":"lt",
+            "ColumnType":"Integer",
+            "ColumnName":"ss_sold_time_sk",
+            "Value":"28855"
+            },
+            "RightNode":{
+            "FilterTypeName":"gt",
+            "ColumnType":"Integer",
+            "ColumnName":"ss_item_sk",
+            "Value":"3"
+            }
+        }
+
+    )";
+
+    int firstRowGroup = 0;
+    int rowGroupToRead = 1;
+
+    ape::Reader test_reader;
+    test_reader.init(file_name, hdfs_host, hdfs_port,requiredColumnName, firstRowGroup, rowGroupToRead);
+    test_reader.setFilter(filterJson);
+
+    bool res = test_reader.doPredicateFilter(0);
+    std::cout<<"predicateFiltere: "<<res<<std::endl;
+
+    test_reader.close();
+
+}

oap-ape/ape-native/src/utils/PredicateExpression.h

@@ -0,0 +1,186 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <parquet/types.h>
+#include <parquet/api/reader.h>
+
+#include "src/utils/Expression.h"
+#include "src/utils/Type.h"
+
+namespace ape {
+
+class PredicateExpression : public Expression {
+ public:
+  explicit PredicateExpression(std::string type_);
+  virtual void Execute() {}
+  int ExecuteWithParam(int batchSize, const std::vector<int64_t>& dataBuffers,
+                               const std::vector<int64_t>& nullBuffers,
+                               std::vector<int8_t>& outBuffers) {
+    return 0;
+  }
+  void setSchema(std::shared_ptr<std::vector<Schema>> schema_) {}
+  virtual int PredicateWithParam(int8_t& out) = 0;
+  virtual void setStatistic(std::shared_ptr<parquet::RowGroupMetaData> rgMataData_) = 0;
+  ~PredicateExpression();
+};
+
+class RootPredicateExpression : public PredicateExpression {
+ public:
+  RootPredicateExpression(std::string type_, std::shared_ptr<PredicateExpression> child_);
+  void Execute() {}
+  int PredicateWithParam(int8_t& out);
+  ~RootPredicateExpression();
+  void setStatistic(std::shared_ptr<parquet::RowGroupMetaData> rgMataData_){
+    child->setStatistic(rgMataData_);
+  }
+  void setSchema(std::shared_ptr<std::vector<Schema>> schema_) {
+    schema = schema_;
+    child->setSchema(schema);
+  }
+
+ private:
+  std::shared_ptr<PredicateExpression> child;
+};
+
+class NotPredicateExpression : public PredicateExpression {
+ public:
+  NotPredicateExpression(std::string type_, std::shared_ptr<PredicateExpression> child_);
+  void Execute() {}
+  int PredicateWithParam(int8_t& out);
+  void setStatistic(std::shared_ptr<parquet::RowGroupMetaData> rgMataData_){
+    child->setStatistic(rgMataData_);
+  }
+  ~NotPredicateExpression();
+  void setSchema(std::shared_ptr<std::vector<Schema>> schema_) {
+    schema = schema_;
+    child->setSchema(schema);
+  }
+  std::shared_ptr<PredicateExpression> getChild() { return child; }
+
+ private:
+  std::shared_ptr<PredicateExpression> child;
+};
+
+class BinaryPredicateExpression : public PredicateExpression {
+ public:
+  BinaryPredicateExpression(std::string type_, std::shared_ptr<PredicateExpression> left_,
+                         std::shared_ptr<PredicateExpression> right_);
+  void Execute() {}
+  int PredicateWithParam(int8_t& out);
+  ~BinaryPredicateExpression();
+  void setSchema(std::shared_ptr<std::vector<Schema>> schema_) {
+    schema = schema_;
+    left->setSchema(schema);
+    right->setSchema(schema);
+  }
+  void setStatistic(std::shared_ptr<parquet::RowGroupMetaData> rgMataData_){
+    left->setStatistic(rgMataData_);
+    right->setStatistic(rgMataData_);
+  }
+  std::shared_ptr<PredicateExpression> getLeftChild() { return left; }
+  std::shared_ptr<PredicateExpression> getRightChild() { return right; }
+
+ private:
+  std::shared_ptr<PredicateExpression> left;
+  std::shared_ptr<PredicateExpression> right;
+  std::string opType;
+};
+
+class UnaryPredicateExpression : public PredicateExpression {
+ public:
+  UnaryPredicateExpression(std::string type_, std::string columnName_)
+      : PredicateExpression(type_) {
+    columnName = columnName_;
+  }
+  ~UnaryPredicateExpression() {}
+  std::string getColumnName();
+
+ protected:
+  std::string columnName;
+};
+
+template <typename T>
+class TypedUnaryPredicateExpression : public UnaryPredicateExpression {
+ public:
+  TypedUnaryPredicateExpression(std::string type_, std::string columnName_, T value_);
+  void Execute() {}
+  int PredicateWithParam(int8_t& out);
+  ~TypedUnaryPredicateExpression();
+  void setSchema(std::shared_ptr<std::vector<Schema>> schema_);
+  void setStatistic(std::shared_ptr<parquet::RowGroupMetaData> rgMataData_);
+
+ private:
+  std::string compareType;
+  T value;
+  T minVal;
+  T maxVal;
+  bool hasMinMax;
+  int columnIndex;
+};
+
+class StringPredicateExpression : public UnaryPredicateExpression {
+ public:
+  StringPredicateExpression(std::string type_, std::string columnName_, std::string value_);
+  ~StringPredicateExpression() {}
+  void setSchema(std::shared_ptr<std::vector<Schema>> schema_);
+  void setStatistic(std::shared_ptr<parquet::RowGroupMetaData> rgMataData_) {}
+  int PredicateWithParam(int8_t& out) = 0;
+  std::string getColumnName();
+
+ protected:
+  std::string type;
+  std::string value;
+  int columnIndex;
+};
+
+class StartWithPredicateExpression : public StringPredicateExpression {
+ public:
+  StartWithPredicateExpression(std::string type_, std::string columnName_,
+                            std::string value_)
+      : StringPredicateExpression(type_, columnName_, value_) {}
+  int PredicateWithParam(int8_t& out);
+  ~StartWithPredicateExpression() {}
+};
+
+class EndWithPredicateExpression : public StringPredicateExpression {
+ public:
+  EndWithPredicateExpression(std::string type_, std::string columnName_, std::string value_)
+      : StringPredicateExpression(type_, columnName_, value_) {}
+  int PredicateWithParam(int8_t& out);
+  ~EndWithPredicateExpression() {}
+};
+
+class ContainsPredicateExpression : public StringPredicateExpression {
+ public:
+  ContainsPredicateExpression(std::string type_, std::string columnName_, std::string value_)
+      : StringPredicateExpression(type_, columnName_, value_) {}
+  int PredicateWithParam(int8_t& out);
+  ~ContainsPredicateExpression() {}
+};
+
+using BoolUnaryPredicateExpression = TypedUnaryPredicateExpression<bool>;
+using Int32UnaryPredicateExpression = TypedUnaryPredicateExpression<int32_t>;
+using Int64UnaryPredicateExpression = TypedUnaryPredicateExpression<int64_t>;
+// using Int96UnaryPredicateExpression = TypedUnaryExpression<int96_t>;
+using FloatUnaryPredicateExpression = TypedUnaryPredicateExpression<float>;
+using DoubleUnaryPredicateExpression = TypedUnaryPredicateExpression<double>;
+using NullUnaryPredicateExpression = TypedUnaryPredicateExpression<NullStruct>;
+using ByteArrayUnaryPredicateExpression = TypedUnaryPredicateExpression<parquet::ByteArray>;
+
+}  // namespace ape