Parquet file reading in VeloxIn10Minutes code

[Manoj-red-hat]

Hi Everyone,

I have created a TPCHq5 simplified query just joining lineitem and orders table.

Just wondering is it possible to read directly generated parquet files rather than generating in place via TPCH connector

    core::PlanNodeId ordersScanId, lineitemScanId;
    auto plan =
        PlanBuilder(planNodeIdGenerator)
            .tpchTableScan(
                tpch::Table::TBL_ORDERS,
                {"o_orderkey", "o_orderdate"},
                2 
                )
            .capturePlanNodeId(ordersScanId)
            .filter(
                "o_orderdate >= DATE '1994-01-01' AND o_orderdate < DATE '1995-01-01'")
            .hashJoin(
                {"o_orderkey"},
                {"l_orderkey"},
                PlanBuilder(planNodeIdGenerator)
                    .tpchTableScan(
                        tpch::Table::TBL_LINEITEM,
                        {"l_orderkey", "l_extendedprice", "l_discount"},
                        2 
                        )
                    .capturePlanNodeId(lineitemScanId)
                    .planNode(),
                "", 
                {"l_extendedprice", "l_discount"})
            .project({"l_extendedprice * (1.0 - l_discount) AS revenue_part"})
            .singleAggregation(
                {}, 
                {"sum(revenue_part) AS revenue"})
            .planNode();
    auto revenueResults = AssertQueryBuilder(plan)
                              .split(ordersScanId, makeTpchSplit())
                              .split(lineitemScanId, makeTpchSplit())
                              .copyResults(pool());
    std::cout << std::endl
              << "> Revenue volume for orders from the year 1994: "
              << revenueResults->toString() << std::endl;
    std::cout << revenueResults->toString(0, 10) << std::endl;

[aditi-pandit]

Hi @Manoj-red-hat,

Thanks for your question and for trying Velox.

The TPC-H connector is for an in-memory generation of Velox vectors using dbgen program. An option for you is to write this generated data to Parquet files, and subsequently read those files in the tpch query plan.

In fact we have implemented with this approach in ParquetTpchTest tests. e.g. Please take a look at
https://github.com/facebookincubator/velox/blob/main/velox/dwio/parquet/tests/ParquetTpchTest.cpp#L81 for more details.

Hope this helps.

[manoj-kumar-tg]

Hi @aditi-pandit ,

was wondering if this is possible, I think there is support for parquet reading in table scan, but somehow its not working for me

  plan = PlanBuilder()
             .tableScan(
                 tpch::Table::TBL_REGION,
                 {"r_nationkey", "r_name","r_comment"},
                 "Parquet",
                 "/home/zetta/part-00001-2c7c4109-2cc1-40d8-8bf4-1435ae57e825-c000.snappy.parquet"
                 )
             .planNode();

  auto regions =
      AssertQueryBuilder(plan).split(makeTpchSplit()).copyResults(pool());

[aditi-pandit]

@manoj-kumar-tg : PlanBuilder.tableScan(...) is for tables associated with HiveConnector. TpchConnector is a different connector in Velox. You will have to use PlanBuilder::tpchTableScan(...) for it. So it seems that you are mixing them in the code above.

Can you give more implementation details for makeTpchSlit() code here ? That would give more hints on the errors.

[Manoj-red-hat]

I want to focus on Velox operators, try to profile and analyze them how they perform, want zero dependency on Duck dB

[Manoj-red-hat]

Something like this
TEST_F(ParquetTableScanTest, countStar) {

[Manoj-red-hat]

@aditi-pandit I want something like this, rather than countstar I just want to run my query which I had mentioned above

[Manoj-red-hat]

@aditi-pandit is this correct
include

   #include "velox/connectors/hive/HiveConnector.h"
   #include "velox/connectors/hive/HiveConnectorSplit.h"

register hive connector

   VeloxIn10MinDemo() {
     // Register Hive connector.
     auto hiveConnector = connector::getConnectorFactory(
         connector::hive::HiveConnectorFactory::kHiveConnectorName)
         ->newConnector("test-hive", std::make_shared<core::MemConfig>());
     connector::registerConnector(hiveConnector);
   }

   ~VeloxIn10MinDemo() {
     connector::unregisterConnector("test-hive");
   }

Create a function to generate a HiveConnectorSplit for a Parquet file:

   std::shared_ptr<connector::hive::HiveConnectorSplit> createParquetSplit(
       const std::string& filePath) {
     return std::make_shared<connector::hive::HiveConnectorSplit>(
         "test-hive",
         filePath,
         dwio::common::FileFormat::PARQUET,
         0, // start
         std::numeric_limits<uint64_t>::max(), // length
         std::unordered_map<std::string, std::string>{}, // partition keys
         nullptr); // sub-split
   }

Use the split in a query plan:

   void VeloxIn10MinDemo::run() {
     // Create a split for the Parquet file.
     auto split = createParquetSplit("/path/to/your/file.parquet");

     // Define the schema of the Parquet file.
     auto rowType = ROW({"column1", "column2"}, {BIGINT(), VARCHAR()});

     // Create a plan to read the Parquet file.
     auto plan = PlanBuilder()
                     .tableScan(rowType)
                     .planNode();

     // Execute the plan.
     auto result = AssertQueryBuilder(plan)
                       .split(split)
                       .copyResults(pool());

     // Print the results.
     std::cout << "> Results from Parquet file: " << result->toString() << std::endl;
     std::cout << result->toString(0, result->size()) << std::endl;
   }

[aditi-pandit]

@Manoj-red-hat : Yes, above code looks more correct to read from Parquet file.