feat: initial PoC Substrait consumer

[davisusanibar]

PoC Java Substrait consumer

C++ Process:

arrow::Status execute_substraitv3() {
    std::cout << "Hello Substrait!" << std::endl;
    ARROW_ASSIGN_OR_RAISE(std::string substrait_json, GetSubstraitJSON());
    std::shared_ptr<arrow::Buffer> shared_buffer = arrow::engine::SerializeJsonPlan(substrait_json).ValueOrDie();
    Result<std::shared_ptr<RecordBatchReader>> reader = arrow::engine::ExecuteSerializedPlan(*shared_buffer);
    struct ArrowArrayStream c_stream;
    arrow::ExportRecordBatchReader(reader.ValueOrDie(), &c_stream);


    // recover values
    std::shared_ptr<RecordBatchReader> new_shared_table = ImportRecordBatchReader(&c_stream).ValueOrDie();
    Result<std::shared_ptr<Table>> new_table = arrow::Table::FromRecordBatchReader(new_shared_table.get());
    std::shared_ptr<Table> new_shared_table_final = new_table.ValueOrDie();
    std::cout << "Values recovered: " << new_shared_table_final->num_rows() << " rows and "
              << new_shared_table_final->num_columns() << " columns" << std::endl;;
    // It prints: Values recovered: 12 rows and 5 columns
}

Java side (Java --> JNI --> C++)

Need to review why on Java side why ArrowReader is populated with column names but not with any data value information.

  @Test
  public void testBaseSubstraitRead() throws Exception {
    try (ArrowArrayStream arrowArrayStream = ArrowArrayStream.allocateNew(rootAllocator())) {
      if (!org.apache.arrow.dataset.substrait.JniWrapper.get().executeSerializedPlan(getSubstraitPlan(), arrowArrayStream.memoryAddress())) {
        System.out.println("No hay nada que mostrar!!!");
      }
      try (ArrowReader arrowReader = Data.importArrayStream(rootAllocator(), arrowArrayStream)){
        System.out.println(arrowReader.getVectorSchemaRoot().contentToTSVString());
        // It prints: Only columns name
        // foo     __fragment_index        __batch_index   __last_in_fragment      __filename
        System.out.println(arrowReader.getVectorSchemaRoot().getSchema());
        // It prints: Schema<foo: Binary, __fragment_index: Int(32, true), __batch_index: Int(32, true), __last_in_fragment: Bool, __filename: Utf8>
      }
    }
  }

[github-actions]

Thanks for opening a pull request!

If this is not a minor PR. Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose

Opening GitHub issues ahead of time contributes to the Openness of the Apache Arrow project.

Then could you also rename the pull request title in the following format?

GH-${GITHUB_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}

or

MINOR: [${COMPONENT}] ${SUMMARY}

In the case of PARQUET issues on JIRA the title also supports:

PARQUET-${JIRA_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}

See also:

• Other pull requests
• Contribution Guidelines - How to contribute patches

[davisusanibar]

PoC Java Substrait consumer

C++ Process:

arrow::Status execute_substraitv3() {
    std::cout << "Hello Substrait!" << std::endl;
    ARROW_ASSIGN_OR_RAISE(std::string substrait_json, GetSubstraitJSON());
    std::shared_ptr<arrow::Buffer> shared_buffer = arrow::engine::SerializeJsonPlan(substrait_json).ValueOrDie();
    Result<std::shared_ptr<RecordBatchReader>> reader = arrow::engine::ExecuteSerializedPlan(*shared_buffer);
    struct ArrowArrayStream c_stream;
    arrow::ExportRecordBatchReader(reader.ValueOrDie(), &c_stream);


    // recover values
    std::shared_ptr<RecordBatchReader> new_shared_table = ImportRecordBatchReader(&c_stream).ValueOrDie();
    Result<std::shared_ptr<Table>> new_table = arrow::Table::FromRecordBatchReader(new_shared_table.get());
    std::shared_ptr<Table> new_shared_table_final = new_table.ValueOrDie();
    std::cout << "Values recovered: " << new_shared_table_final->num_rows() << " rows and "
              << new_shared_table_final->num_columns() << " columns" << std::endl;;
    // It prints: Values recovered: 12 rows and 5 columns
}

Java side (Java --> JNI --> C++)

Need to review why on Java side why ArrowReader is populated with column names but not with any data value information.

  @Test
  public void testBaseSubstraitRead() throws Exception {
    try (ArrowArrayStream arrowArrayStream = ArrowArrayStream.allocateNew(rootAllocator())) {
      if (!org.apache.arrow.dataset.substrait.JniWrapper.get().executeSerializedPlan(getSubstraitPlan(), arrowArrayStream.memoryAddress())) {
        System.out.println("No hay nada que mostrar!!!");
      }
      try (ArrowReader arrowReader = Data.importArrayStream(rootAllocator(), arrowArrayStream)){
        System.out.println(arrowReader.getVectorSchemaRoot().contentToTSVString());
        // It prints: Only columns name
        // foo     __fragment_index        __batch_index   __last_in_fragment      __filename
        System.out.println(arrowReader.getVectorSchemaRoot().getSchema());
        // It prints: Schema<foo: Binary, __fragment_index: Int(32, true), __batch_index: Int(32, true), __last_in_fragment: Bool, __filename: Utf8>
      }
    }
  }

Problem was that client was not calling arrowReader.loadNextBatch() to start reading data.

[sonatype-lift]

SC2223: This default assignment may cause DoS due to globbing. Quote it.

---

ℹ️ Expand to see all @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command	Usage
@sonatype-lift ignore	Leave out the above finding from this PR
@sonatype-lift ignoreall	Leave out all the existing findings from this PR
@sonatype-lift exclude <file|issue|path|tool>	Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

---

Help us improve LIFT! (Sonatype LiftBot external survey)

Was this a good recommendation for you? _{Answering this survey will not impact your Lift settings.}

[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]