You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "lucasfang (via GitHub)" <gi...@apache.org> on 2026/06/25 06:29:43 UTC
[PR] feat: add ORC format reader/writer utilities [paimon-cpp]
lucasfang opened a new pull request, #122:
URL: https://github.com/apache/paimon-cpp/pull/122
### Purpose
Linked issue: No linked issue
This change adds ORC format reader and writer utilities for batch processing Paimon data files.
Included changes:
- **ORC File Batch Reader**:
- Adds `OrcFileBatchReader` header and implementation for batch reading ORC files
- Adds comprehensive test coverage in `orc_file_batch_reader_test.cpp`
- **ORC Format Writer**:
- Adds `OrcFormatWriter` header and implementation for writing ORC format files
- Adds test coverage in `orc_format_writer_test.cpp`
- **ORC Reader Wrapper**:
- Adds `OrcReaderWrapper` header and implementation as a wrapper utility for ORC readers
- Adds test coverage in `orc_reader_wrapper_test.cpp`
### Tests
Not run. Local compile, CMake, and gtest environment checks are not part of this PR description.
Test coverage included in this change:
- `OrcFileBatchReaderTest`
- `OrcFormatWriterTest`
- `OrcReaderWrapperTest`
### API and Format
No public API, storage format, or protocol changes.
All added files are under `src/paimon/format/orc/` (internal implementation). No headers added under `include/`.
### Documentation
No documentation changes required.
### Generative AI tooling
Migrate-by: Aone Copilot (Qwen3.7-Max)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: add ORC format reader/writer utilities [paimon-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #122:
URL: https://github.com/apache/paimon-cpp/pull/122#discussion_r3472364033
##########
src/paimon/format/orc/orc_reader_wrapper.cpp:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.
+ */
+
+#include "paimon/format/orc/orc_reader_wrapper.h"
+
+#include <algorithm>
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "fmt/format.h"
+#include "orc/OrcFile.hh"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/scope_guard.h"
+
+namespace paimon::orc {
+
+Status OrcReaderWrapper::SeekToRow(uint64_t row_number) {
+ try {
+ row_reader_->seekToRow(row_number);
+ next_row_ = row_number;
+ } catch (const std::exception& e) {
+ return Status::Invalid(
+ fmt::format("orc file batch reader seek to row {} failed for file {}, with {} error",
+ row_number, file_name_, e.what()));
+ } catch (...) {
+ return Status::UnknownError(fmt::format(
+ "orc file batch reader seek to row {} failed for file {}, with unknown error",
+ row_number, file_name_));
+ }
+ return Status::OK();
+}
+
+Status OrcReaderWrapper::SetReadSchema(const std::shared_ptr<arrow::DataType>& target_type,
+ const ::orc::RowReaderOptions& row_reader_options) {
+ try {
+ row_reader_ = reader_->createRowReader(row_reader_options);
+ target_type_ = target_type;
+ } catch (const std::exception& e) {
+ return Status::Invalid(
+ fmt::format("orc file batch reader create row reader failed for file {}, with {} error",
+ file_name_, e.what()));
+ } catch (...) {
+ return Status::UnknownError(fmt::format(
+ "orc file batch reader create row reader failed for file {}, with unknown error",
+ file_name_));
+ }
+ return Status::OK();
+}
+
+Result<BatchReader::ReadBatch> OrcReaderWrapper::Next() {
+ if (has_error_) {
+ return Status::Invalid(fmt::format(
+ "Since an error has occurred, next batch has been prohibited. file '{}'", file_name_));
+ }
+ std::unique_ptr<ArrowArray> c_array = std::make_unique<ArrowArray>();
+ std::unique_ptr<ArrowSchema> c_schema = std::make_unique<ArrowSchema>();
+ try {
+ auto orc_batch = row_reader_->createRowBatch(batch_size_);
+ bool eof = !row_reader_->next(*orc_batch);
+ if (eof) {
+ return BatchReader::MakeEofBatch();
+ }
+ ScopeGuard guard([this]() { has_error_ = true; });
+ assert(orc_batch->numElements > 0);
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<arrow::Array> array,
+ OrcAdapter::AppendBatch(target_type_, orc_batch.get(), arrow_pool_.get()));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, c_array.get(), c_schema.get()));
+ next_row_ = GetRowNumber() + orc_batch->numElements;
+ guard.Release();
Review Comment:
`next_row_` is updated as `GetRowNumber() + orc_batch->numElements`, but `::orc::RowReader::getRowNumber()` already reports the next row to be read (after `next()` it has advanced by `numElements`). Adding `numElements` again will over-advance the reported position (and breaks `GetNextRowToRead()` consumers/tests).
##########
src/paimon/format/orc/orc_file_batch_reader_test.cpp:
##########
@@ -0,0 +1,750 @@
+/*
+ * 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.
+ */
+
+#include "paimon/format/orc/orc_file_batch_reader.h"
+
+#include <list>
+#include <map>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/c/bridge.h"
+#include "arrow/ipc/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/defs.h"
+#include "paimon/format/orc/orc_adapter.h"
+#include "paimon/format/orc/orc_format_defs.h"
+#include "paimon/format/orc/orc_format_writer.h"
+#include "paimon/format/orc/orc_input_stream_impl.h"
+#include "paimon/format/orc/orc_memory_pool.h"
+#include "paimon/format/orc/orc_metrics.h"
+#include "paimon/format/orc/orc_output_stream_impl.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/predicate/predicate_builder.h"
+#include "paimon/testing/utils/read_result_collector.h"
+#include "paimon/testing/utils/testharness.h"
+#include "paimon/testing/utils/timezone_guard.h"
+
+namespace paimon::orc::test {
+
+struct TestParam {
+ uint64_t natural_read_size;
+ bool enable_tz;
+};
+
+class OrcFileBatchReaderTest : public ::testing::Test,
+ public ::testing::WithParamInterface<TestParam> {
+ public:
+ void SetUp() override {
+ pool_ = GetDefaultPool();
+ batch_size_ = 10;
+
+ arrow::FieldVector fields = {
+ arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
+ arrow::field("f2", arrow::int32()), arrow::field("f3", arrow::float64())};
+
+ struct_array_ = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+ ["Bob", 10, 0, 12.1], ["Emily", 10, 0, 13.1], ["Tony", 10, 0, 14.1], ["Emily", 10, 0, 15.1],
+ ["Bob", 10, 0, 12.1], ["Alex", 10, 0, 16.1], ["David", 10, 0, 17.1], ["Lily", 10, 0, 17.1]
+ ])")
+ .ValueOrDie());
+ }
+ void TearDown() override {}
+
+ std::pair<std::unique_ptr<OrcFileBatchReader>, std::shared_ptr<arrow::ChunkedArray>>
+ ReadBatchWithCustomizedData(const std::shared_ptr<arrow::Array>& src_array,
+ int32_t write_batch_size, int32_t write_stripe_size,
+ int32_t write_row_index_stride, const arrow::Schema* read_schema,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::optional<RoaringBitmap32>& selection_bitmap,
+ int32_t read_batch_size, double dict_key_size_threshold,
+ bool enable_lazy_decoding) const {
+ arrow::Schema src_schema(src_array->type()->fields());
+ EXPECT_OK_AND_ASSIGN(std::unique_ptr<::orc::Type> orc_type,
+ OrcAdapter::GetOrcType(src_schema));
+ EXPECT_TRUE(orc_type);
+ auto dir = paimon::test::UniqueTestDirectory::Create();
+ EXPECT_TRUE(dir);
+ auto fs = dir->GetFileSystem();
+ std::string data_path = dir->Str() + "/test.data";
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<paimon::OutputStream> output_stream,
+ fs->Create(data_path, /*overwrite=*/true));
+ EXPECT_TRUE(output_stream);
+ EXPECT_OK_AND_ASSIGN(auto orc_output_stream, OrcOutputStreamImpl::Create(output_stream));
+ EXPECT_TRUE(orc_output_stream);
+ ::orc::WriterOptions writer_options;
+ writer_options.setDictionaryKeySizeThreshold(dict_key_size_threshold);
+ if (write_stripe_size != -1) {
+ writer_options.setStripeSize(write_stripe_size);
+ }
+ if (write_row_index_stride != -1) {
+ writer_options.setRowIndexStride(write_row_index_stride);
+ }
+ std::unique_ptr<::orc::Writer> writer =
+ ::orc::createWriter(*orc_type, orc_output_stream.get(), writer_options);
+ // for simple case, assume that src_array.length() % write_batch_size == 0
+ EXPECT_EQ(src_array->length() % write_batch_size, 0);
+ int32_t write_batch_count = src_array->length() / write_batch_size;
+ for (int32_t i = 0; i < write_batch_count; i++) {
+ auto src_slice = src_array->Slice(i * write_batch_size, write_batch_size);
+ auto write_batch = writer->createRowBatch(src_slice->length());
+ // Convert from arrow array to orc batch
+ EXPECT_OK(OrcAdapter::WriteBatch(src_slice, write_batch.get()));
+ writer->add(*write_batch);
+ }
+ writer->close();
+ EXPECT_OK(output_stream->Close());
+
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<paimon::InputStream> input_stream,
+ fs->Open(data_path));
+ EXPECT_TRUE(input_stream);
+ EXPECT_OK_AND_ASSIGN(auto orc_input_stream,
+ OrcInputStreamImpl::Create(input_stream, DEFAULT_NATURAL_READ_SIZE));
+ EXPECT_TRUE(orc_input_stream);
+ std::string enable_lazy_decoding_str = enable_lazy_decoding ? "true" : "false";
+ std::map<std::string, std::string> options = {
+ {ORC_READ_ENABLE_LAZY_DECODING, enable_lazy_decoding_str}};
+ auto orc_batch_reader =
+ PrepareOrcFileBatchReader(std::move(orc_input_stream), options, read_schema, predicate,
+ selection_bitmap, read_batch_size);
+ EXPECT_OK_AND_ASSIGN(
+ auto result, paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ return std::make_pair(std::move(orc_batch_reader), result);
+ }
+
+ std::unique_ptr<OrcFileBatchReader> PrepareOrcFileBatchReader(
+ const std::string& file_name, const arrow::Schema* read_schema, int32_t batch_size,
+ uint64_t natural_read_size) const {
+ std::shared_ptr<FileSystem> file_system = std::make_shared<LocalFileSystem>();
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input_stream,
+ file_system->Open(file_name));
+ EXPECT_TRUE(input_stream);
+ EXPECT_OK_AND_ASSIGN(auto in_stream,
+ OrcInputStreamImpl::Create(input_stream, natural_read_size));
+ EXPECT_TRUE(in_stream);
+ std::map<std::string, std::string> options = {{ORC_READ_ENABLE_LAZY_DECODING, "true"},
+ {ORC_READ_ENABLE_METRICS, "true"},
+ {"orc.timestamp-ltz.legacy.type", "false"}};
+ return PrepareOrcFileBatchReader(std::move(in_stream), options, read_schema,
+ /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt,
+ batch_size);
+ }
+
+ std::unique_ptr<paimon::orc::OrcFileBatchReader> PrepareOrcFileBatchReader(
+ std::unique_ptr<::orc::InputStream>&& in_stream,
+ const std::map<std::string, std::string>& options, const arrow::Schema* read_schema,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::optional<RoaringBitmap32>& selection_bitmap, int32_t batch_size) const {
+ EXPECT_OK_AND_ASSIGN(
+ auto orc_batch_reader,
+ OrcFileBatchReader::Create(std::move(in_stream), pool_, options, batch_size))
+ EXPECT_TRUE(orc_batch_reader);
Review Comment:
The `EXPECT_OK_AND_ASSIGN(...)` statement is missing a terminating semicolon, which will prevent this test file from compiling.
##########
src/paimon/format/orc/orc_file_batch_reader_test.cpp:
##########
@@ -0,0 +1,750 @@
+/*
+ * 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.
+ */
+
+#include "paimon/format/orc/orc_file_batch_reader.h"
+
+#include <list>
+#include <map>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/c/bridge.h"
+#include "arrow/ipc/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/defs.h"
+#include "paimon/format/orc/orc_adapter.h"
+#include "paimon/format/orc/orc_format_defs.h"
+#include "paimon/format/orc/orc_format_writer.h"
+#include "paimon/format/orc/orc_input_stream_impl.h"
+#include "paimon/format/orc/orc_memory_pool.h"
+#include "paimon/format/orc/orc_metrics.h"
+#include "paimon/format/orc/orc_output_stream_impl.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/predicate/predicate_builder.h"
+#include "paimon/testing/utils/read_result_collector.h"
+#include "paimon/testing/utils/testharness.h"
+#include "paimon/testing/utils/timezone_guard.h"
+
+namespace paimon::orc::test {
+
+struct TestParam {
+ uint64_t natural_read_size;
+ bool enable_tz;
+};
+
+class OrcFileBatchReaderTest : public ::testing::Test,
+ public ::testing::WithParamInterface<TestParam> {
+ public:
+ void SetUp() override {
+ pool_ = GetDefaultPool();
+ batch_size_ = 10;
+
+ arrow::FieldVector fields = {
+ arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
+ arrow::field("f2", arrow::int32()), arrow::field("f3", arrow::float64())};
+
+ struct_array_ = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+ ["Bob", 10, 0, 12.1], ["Emily", 10, 0, 13.1], ["Tony", 10, 0, 14.1], ["Emily", 10, 0, 15.1],
+ ["Bob", 10, 0, 12.1], ["Alex", 10, 0, 16.1], ["David", 10, 0, 17.1], ["Lily", 10, 0, 17.1]
+ ])")
+ .ValueOrDie());
+ }
+ void TearDown() override {}
+
+ std::pair<std::unique_ptr<OrcFileBatchReader>, std::shared_ptr<arrow::ChunkedArray>>
+ ReadBatchWithCustomizedData(const std::shared_ptr<arrow::Array>& src_array,
+ int32_t write_batch_size, int32_t write_stripe_size,
+ int32_t write_row_index_stride, const arrow::Schema* read_schema,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::optional<RoaringBitmap32>& selection_bitmap,
+ int32_t read_batch_size, double dict_key_size_threshold,
+ bool enable_lazy_decoding) const {
+ arrow::Schema src_schema(src_array->type()->fields());
+ EXPECT_OK_AND_ASSIGN(std::unique_ptr<::orc::Type> orc_type,
+ OrcAdapter::GetOrcType(src_schema));
+ EXPECT_TRUE(orc_type);
+ auto dir = paimon::test::UniqueTestDirectory::Create();
+ EXPECT_TRUE(dir);
+ auto fs = dir->GetFileSystem();
+ std::string data_path = dir->Str() + "/test.data";
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<paimon::OutputStream> output_stream,
+ fs->Create(data_path, /*overwrite=*/true));
+ EXPECT_TRUE(output_stream);
+ EXPECT_OK_AND_ASSIGN(auto orc_output_stream, OrcOutputStreamImpl::Create(output_stream));
+ EXPECT_TRUE(orc_output_stream);
+ ::orc::WriterOptions writer_options;
+ writer_options.setDictionaryKeySizeThreshold(dict_key_size_threshold);
+ if (write_stripe_size != -1) {
+ writer_options.setStripeSize(write_stripe_size);
+ }
+ if (write_row_index_stride != -1) {
+ writer_options.setRowIndexStride(write_row_index_stride);
+ }
+ std::unique_ptr<::orc::Writer> writer =
+ ::orc::createWriter(*orc_type, orc_output_stream.get(), writer_options);
+ // for simple case, assume that src_array.length() % write_batch_size == 0
+ EXPECT_EQ(src_array->length() % write_batch_size, 0);
+ int32_t write_batch_count = src_array->length() / write_batch_size;
+ for (int32_t i = 0; i < write_batch_count; i++) {
+ auto src_slice = src_array->Slice(i * write_batch_size, write_batch_size);
+ auto write_batch = writer->createRowBatch(src_slice->length());
+ // Convert from arrow array to orc batch
+ EXPECT_OK(OrcAdapter::WriteBatch(src_slice, write_batch.get()));
+ writer->add(*write_batch);
+ }
+ writer->close();
+ EXPECT_OK(output_stream->Close());
+
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<paimon::InputStream> input_stream,
+ fs->Open(data_path));
+ EXPECT_TRUE(input_stream);
+ EXPECT_OK_AND_ASSIGN(auto orc_input_stream,
+ OrcInputStreamImpl::Create(input_stream, DEFAULT_NATURAL_READ_SIZE));
+ EXPECT_TRUE(orc_input_stream);
+ std::string enable_lazy_decoding_str = enable_lazy_decoding ? "true" : "false";
+ std::map<std::string, std::string> options = {
+ {ORC_READ_ENABLE_LAZY_DECODING, enable_lazy_decoding_str}};
+ auto orc_batch_reader =
+ PrepareOrcFileBatchReader(std::move(orc_input_stream), options, read_schema, predicate,
+ selection_bitmap, read_batch_size);
+ EXPECT_OK_AND_ASSIGN(
+ auto result, paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ return std::make_pair(std::move(orc_batch_reader), result);
+ }
+
+ std::unique_ptr<OrcFileBatchReader> PrepareOrcFileBatchReader(
+ const std::string& file_name, const arrow::Schema* read_schema, int32_t batch_size,
+ uint64_t natural_read_size) const {
+ std::shared_ptr<FileSystem> file_system = std::make_shared<LocalFileSystem>();
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input_stream,
+ file_system->Open(file_name));
+ EXPECT_TRUE(input_stream);
+ EXPECT_OK_AND_ASSIGN(auto in_stream,
+ OrcInputStreamImpl::Create(input_stream, natural_read_size));
+ EXPECT_TRUE(in_stream);
+ std::map<std::string, std::string> options = {{ORC_READ_ENABLE_LAZY_DECODING, "true"},
+ {ORC_READ_ENABLE_METRICS, "true"},
+ {"orc.timestamp-ltz.legacy.type", "false"}};
+ return PrepareOrcFileBatchReader(std::move(in_stream), options, read_schema,
+ /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt,
+ batch_size);
+ }
+
+ std::unique_ptr<paimon::orc::OrcFileBatchReader> PrepareOrcFileBatchReader(
+ std::unique_ptr<::orc::InputStream>&& in_stream,
+ const std::map<std::string, std::string>& options, const arrow::Schema* read_schema,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::optional<RoaringBitmap32>& selection_bitmap, int32_t batch_size) const {
+ EXPECT_OK_AND_ASSIGN(
+ auto orc_batch_reader,
+ OrcFileBatchReader::Create(std::move(in_stream), pool_, options, batch_size))
+ EXPECT_TRUE(orc_batch_reader);
+ std::unique_ptr<ArrowSchema> c_schema = std::make_unique<ArrowSchema>();
+ auto arrow_status = arrow::ExportSchema(*read_schema, c_schema.get());
+ EXPECT_TRUE(arrow_status.ok());
+ EXPECT_OK(orc_batch_reader->SetReadSchema(c_schema.get(), predicate, selection_bitmap));
+ return orc_batch_reader;
+ }
+
+ void WriteArray(const std::shared_ptr<FileSystem>& fs, const std::string& file_path,
+ const std::shared_ptr<arrow::Array>& src_array,
+ const std::shared_ptr<arrow::Schema>& arrow_schema,
+ const std::map<std::string, std::string>& options) const {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> out,
+ fs->Create(file_path, /*overwrite=*/true));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OrcOutputStreamImpl> output_stream,
+ OrcOutputStreamImpl::Create(out));
+ ASSERT_OK_AND_ASSIGN(
+ auto format_writer,
+ OrcFormatWriter::Create(std::move(output_stream), *arrow_schema, options, "zstd",
+ /*batch_size=*/src_array->length(), pool_));
+ auto arrow_array = std::make_unique<ArrowArray>();
+ ASSERT_TRUE(arrow::ExportArray(*src_array, arrow_array.get()).ok());
+ ASSERT_OK(format_writer->AddBatch(arrow_array.get()));
+ ASSERT_OK(format_writer->Flush());
+ ASSERT_OK(format_writer->Finish());
+ ASSERT_OK(out->Flush());
+ ASSERT_OK(out->Close());
+ }
+
+ private:
+ std::shared_ptr<MemoryPool> pool_;
+ int32_t batch_size_;
+ std::shared_ptr<arrow::StructArray> struct_array_;
+};
+
+INSTANTIATE_TEST_SUITE_P(TestParam, OrcFileBatchReaderTest,
+ ::testing::Values(TestParam{128 * 1024, false}, TestParam{16, false},
+ TestParam{16, true}));
+
+TEST_F(OrcFileBatchReaderTest, TestSetReadSchema) {
+ std::string file_name = paimon::test::GetDataDir() +
+ "/orc/append_09.db/append_09/f1=10/bucket-1/"
+ "data-b9e7c41f-66e8-4dad-b25a-e6e1963becc4-0.orc";
+ std::shared_ptr<FileSystem> file_system = std::make_shared<LocalFileSystem>();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input_stream, file_system->Open(file_name));
+ ASSERT_OK_AND_ASSIGN(auto in_stream,
+ OrcInputStreamImpl::Create(input_stream, DEFAULT_NATURAL_READ_SIZE));
+ std::map<std::string, std::string> options = {{ORC_READ_ENABLE_LAZY_DECODING, "true"}};
+ ASSERT_OK_AND_ASSIGN(
+ auto orc_batch_reader,
+ OrcFileBatchReader::Create(std::move(in_stream), pool_, options, batch_size_));
+ // test GetFileSchema()
+ ASSERT_OK_AND_ASSIGN(auto c_file_schema, orc_batch_reader->GetFileSchema());
+ auto arrow_file_schema = arrow::ImportSchema(c_file_schema.get()).ValueOrDie();
+ arrow::FieldVector fields = {
+ arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
+ arrow::field("f2", arrow::int32()), arrow::field("f3", arrow::float64())};
+ arrow::Schema file_schema(fields);
+ ASSERT_TRUE(arrow_file_schema->Equals(file_schema));
+ // NextBatch() without SetReadSchema(), will return data with
+ // file schema
+ ASSERT_OK_AND_ASSIGN(auto result_with_file_schema,
+ paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ auto expected_file_chunk_array = std::make_shared<arrow::ChunkedArray>(struct_array_);
+ ASSERT_TRUE(result_with_file_schema->Equals(expected_file_chunk_array));
+ // NextBatch() with SetReadSchema(), will return data with read schema
+ arrow::Schema read_schema({fields[0], fields[2]});
+ std::unique_ptr<ArrowSchema> c_read_schema = std::make_unique<ArrowSchema>();
+ ASSERT_TRUE(arrow::ExportSchema(read_schema, c_read_schema.get()).ok());
+ ASSERT_OK(orc_batch_reader->SetReadSchema(c_read_schema.get(),
+ /*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt));
+ auto expected_read_array =
+ arrow::StructArray::Make({struct_array_->field(0), struct_array_->field(2)},
+ {fields[0], fields[2]})
+ .ValueOrDie();
+ auto expected_read_chunk_array = std::make_shared<arrow::ChunkedArray>(expected_read_array);
+ ASSERT_OK_AND_ASSIGN(auto result_with_read_schema,
+ paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ ASSERT_TRUE(result_with_read_schema->Equals(expected_read_chunk_array));
+
+ // NextBatch() with predicate
+ auto predicate = PredicateBuilder::GreaterThan(
+ /*field_index=*/3, /*field_name=*/"f2", FieldType::INT, Literal(2));
+ c_read_schema = std::make_unique<ArrowSchema>();
+ ASSERT_TRUE(arrow::ExportSchema(read_schema, c_read_schema.get()).ok());
+ ASSERT_OK(orc_batch_reader->SetReadSchema(c_read_schema.get(), predicate,
+ /*selection_bitmap=*/std::nullopt));
+ ASSERT_OK_AND_ASSIGN(result_with_read_schema,
+ paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ ASSERT_FALSE(result_with_read_schema);
+}
+
+TEST_F(OrcFileBatchReaderTest, TestCreateRowReaderOptions) {
+ auto orc_pool = std::make_shared<OrcMemoryPool>(pool_);
+ std::vector<uint64_t> target_column_ids;
+ {
+ // read all fields && default options
+ std::map<std::string, std::string> options;
+ std::string orc_schema = "struct<col1:int,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(orc_schema);
+ std::unique_ptr<::orc::Type> target_type = ::orc::Type::buildTypeFromString(orc_schema);
+ ASSERT_OK_AND_ASSIGN(auto row_reader_option,
+ OrcFileBatchReader::CreateRowReaderOptions(
+ src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options, &target_column_ids));
+ ASSERT_EQ(std::list<std::string>({"col1", "col2", "col3"}),
+ row_reader_option.getIncludeNames());
+ ASSERT_EQ(row_reader_option.getEnableLazyDecoding(), false);
+ }
+ {
+ // read partial fields && set options
+ std::map<std::string, std::string> options = {{ORC_READ_ENABLE_LAZY_DECODING, "true"}};
+ std::string src_orc_schema = "struct<col1:int,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(src_orc_schema);
+ std::string target_orc_schema = "struct<col1:int,col3:string>";
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString(target_orc_schema);
+ ASSERT_OK_AND_ASSIGN(auto row_reader_option,
+ OrcFileBatchReader::CreateRowReaderOptions(
+ src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options, &target_column_ids));
+ ASSERT_EQ(std::list<std::string>({"col1", "col3"}), row_reader_option.getIncludeNames());
+ ASSERT_EQ(row_reader_option.getEnableLazyDecoding(), true);
+ }
+ {
+ // read partial fields, sequence mismatch
+ std::map<std::string, std::string> options;
+ std::string src_orc_schema = "struct<col1:int,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(src_orc_schema);
+ std::string target_orc_schema = "struct<col3:string,col1:int>";
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString(target_orc_schema);
+ ASSERT_NOK_WITH_MSG(
+ OrcFileBatchReader::CreateRowReaderOptions(src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options,
+ &target_column_ids),
+ "The column id of the target field should be monotonically increasing in format "
+ "reader");
+ }
+ {
+ // read non exist column
+ std::map<std::string, std::string> options;
+ std::string src_orc_schema = "struct<col1:int,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(src_orc_schema);
+ std::string target_orc_schema = "struct<col1:int,non_exist_col:double>";
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString(target_orc_schema);
+ ASSERT_NOK_WITH_MSG(OrcFileBatchReader::CreateRowReaderOptions(
+ src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options, &target_column_ids),
+ "field non_exist_col not in file schema");
+ }
+ {
+ std::map<std::string, std::string> options;
+ std::string src_orc_schema =
+ "struct<col1:struct<sub1:int,sub2:array<array<double>>,sub3:int>,col2:double,col3:"
+ "map<string,int>>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(src_orc_schema);
+ std::string target_orc_schema =
+ "struct<col1:struct<sub1:int,sub2:array<array<double>>,sub3:int>,col3:map<string,int>>";
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString(target_orc_schema);
+ ASSERT_OK_AND_ASSIGN(auto row_reader_option,
+ OrcFileBatchReader::CreateRowReaderOptions(
+ src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options, &target_column_ids));
+ ASSERT_EQ(std::list<std::string>({"col1", "col3"}), row_reader_option.getIncludeNames());
+ }
+ {
+ // read with type mismatch
+ std::map<std::string, std::string> options;
+ std::string src_orc_schema =
+ "struct<col1:struct<sub1:int,sub2:int,sub3:int>,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(src_orc_schema);
+ std::string target_orc_schema =
+ "struct<col1:struct<sub1:int,sub2:double,sub3:int>,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString(target_orc_schema);
+
+ ASSERT_NOK_WITH_MSG(
+ OrcFileBatchReader::CreateRowReaderOptions(src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options,
+ &target_column_ids),
+ "target_type "
+ "struct<col1:struct<sub1:int,sub2:double,sub3:int>,col2:double,col3:string> not match "
+ "src_type struct<col1:struct<sub1:int,sub2:int,sub3:int>,col2:double,col3:string>, "
+ "mismatch field name col1");
+ }
+}
+
+TEST_P(OrcFileBatchReaderTest, TestNextBatchSimple) {
+ std::string file_name = paimon::test::GetDataDir() +
+ "/orc/append_09.db/append_09/f1=10/bucket-1/"
+ "data-b9e7c41f-66e8-4dad-b25a-e6e1963becc4-0.orc";
+ auto data_type = struct_array_->struct_type();
+ arrow::Schema read_schema(data_type->fields());
+ auto [natural_read_size, _] = GetParam();
+ for (auto batch_size : {1, 2, 3, 5, 8, 10}) {
+ auto orc_batch_reader =
+ PrepareOrcFileBatchReader(file_name, &read_schema, batch_size, natural_read_size);
+ ASSERT_EQ(orc_batch_reader->GetPreviousBatchFirstRowNumber().value(), -1);
+ ASSERT_OK_AND_ASSIGN(auto result_array, paimon::test::ReadResultCollector::CollectResult(
+ orc_batch_reader.get()));
+ ASSERT_EQ(orc_batch_reader->GetPreviousBatchFirstRowNumber().value(), 8);
+ orc_batch_reader->Close();
+ auto expected_array = std::make_shared<arrow::ChunkedArray>(struct_array_);
+ ASSERT_TRUE(result_array->Equals(expected_array));
+ // test metrics
+ auto reader_metrics = orc_batch_reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t io_count,
+ reader_metrics->GetCounter(OrcMetrics::READ_IO_COUNT));
+ ASSERT_GT(io_count, 0);
+ ASSERT_OK_AND_ASSIGN(uint64_t latency,
+ reader_metrics->GetCounter(OrcMetrics::READ_INCLUSIVE_LATENCY_US));
+ ASSERT_GT(latency, 0);
+ }
+}
+
+TEST_P(OrcFileBatchReaderTest, TestNextBatchWithTargetSchema) {
+ std::string file_name = paimon::test::GetDataDir() +
+ "/orc/append_09.db/append_09/f1=10/bucket-1/"
+ "data-b9e7c41f-66e8-4dad-b25a-e6e1963becc4-0.orc";
+ // read without f2
+ auto data_type = struct_array_->struct_type();
+ arrow::FieldVector fields = {data_type->GetFieldByName("f0"), data_type->GetFieldByName("f1"),
+ data_type->GetFieldByName("f3")};
+ auto [natural_read_size, _] = GetParam();
+ arrow::Schema read_schema(fields);
+
+ auto orc_batch_reader =
+ PrepareOrcFileBatchReader(file_name, &read_schema, batch_size_, natural_read_size);
+ ASSERT_OK_AND_ASSIGN(auto result_array,
+ paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ auto expected_array = std::make_shared<arrow::ChunkedArray>(
+ arrow::StructArray::Make(
+ {struct_array_->GetFieldByName("f0"), struct_array_->GetFieldByName("f1"),
+ struct_array_->GetFieldByName("f3")},
+ fields)
+ .ValueOrDie());
+ ASSERT_TRUE(result_array->Equals(expected_array));
+}
+
+TEST_F(OrcFileBatchReaderTest, TestNextBatchWithOutofOrderTargetSchema) {
+ std::string file_name = paimon::test::GetDataDir() +
+ "/orc/append_09.db/append_09/f1=10/bucket-1/"
+ "data-b9e7c41f-66e8-4dad-b25a-e6e1963becc4-0.orc";
+ // file type is f0, f1, f2, f3
+ // read with f3, f1, f0
+ auto data_type = struct_array_->struct_type();
+ arrow::FieldVector fields = {data_type->GetFieldByName("f3"), data_type->GetFieldByName("f1"),
+ data_type->GetFieldByName("f0")};
+ arrow::Schema read_schema(fields);
+ std::shared_ptr<FileSystem> file_system = std::make_shared<LocalFileSystem>();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input_stream, file_system->Open(file_name));
+ ASSERT_OK_AND_ASSIGN(auto in_stream,
+ OrcInputStreamImpl::Create(input_stream, DEFAULT_NATURAL_READ_SIZE));
+ std::map<std::string, std::string> options = {{ORC_READ_ENABLE_LAZY_DECODING, "true"}};
+ ASSERT_OK_AND_ASSIGN(
+ auto orc_batch_reader,
+ OrcFileBatchReader::Create(std::move(in_stream), pool_, options, batch_size_));
+ std::unique_ptr<ArrowSchema> c_schema = std::make_unique<ArrowSchema>();
+ ASSERT_TRUE(arrow::ExportSchema(read_schema, c_schema.get()).ok());
+ ASSERT_NOK_WITH_MSG(orc_batch_reader->SetReadSchema(c_schema.get(), /*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt),
+ "The column id of the target field should be "
+ "monotonically increasing in "
+ "format reader");
+}
+
+TEST_P(OrcFileBatchReaderTest, TestNextBatchWithNullValue) {
+ std::string file_name = paimon::test::GetDataDir() +
+ "/orc/append_09.db/append_09/f1=20/bucket-0/"
+ "data-b913a160-a4d1-4084-af2a-18333c35668e-0.orc";
+ arrow::FieldVector fields = {
+ arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
+ arrow::field("f2", arrow::int32()), arrow::field("f3", arrow::float64())};
+ auto arrow_data_type = arrow::struct_(fields);
+ arrow::Schema read_schema(fields);
+ auto [natural_read_size, _] = GetParam();
+ auto orc_batch_reader =
+ PrepareOrcFileBatchReader(file_name, &read_schema, batch_size_, natural_read_size);
+ ASSERT_OK_AND_ASSIGN(auto result_array,
+ paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ std::string json = R"([
+ ["Paul", 20, 1, null]
+ ])";
+ auto array_status =
+ arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow_data_type, {json}, &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ ASSERT_TRUE(result_array->Equals(expected_array));
+}
+
+TEST_F(OrcFileBatchReaderTest, TestNextBatchWithDictionary) {
+ auto f0 = arrow::field("f0", arrow::list(arrow::utf8()));
+ auto f1 = arrow::field("f1", arrow::map(arrow::utf8(), arrow::binary()));
+ auto f2 = arrow::field(
+ "f2", arrow::struct_({field("sub1", arrow::int64()), field("sub2", arrow::binary()),
+ field("sub3", arrow::utf8())}));
Review Comment:
Unqualified `field(...)` is used here, but there is no `field` symbol in the current namespace; it should be `arrow::field(...)` like the rest of the file. As-is, this test won't compile.
##########
src/paimon/format/orc/orc_format_writer.h:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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 <cstdint>
+#include <map>
+#include <memory>
+#include <string>
+
+#include "arrow/api.h"
+#include "orc/Common.hh"
+#include "orc/OrcFile.hh"
+#include "orc/Type.hh"
+#include "orc/Vector.hh"
+#include "orc/Writer.hh"
+#include "paimon/format/format_writer.h"
+#include "paimon/format/orc/orc_memory_pool.h"
+#include "paimon/metrics.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace arrow {
+class DataType;
+class Schema;
+} // namespace arrow
+namespace paimon {
+class MemoryPool;
+namespace orc {
+class OrcMemoryPool;
+} // namespace orc
+} // namespace paimon
+struct ArrowArray;
+
+namespace paimon::orc {
+/// A `FormatWriter` implementation that writes data in ORC format.
+class OrcFormatWriter : public FormatWriter {
+ public:
+ static Result<std::unique_ptr<OrcFormatWriter>> Create(
+ std::unique_ptr<::orc::OutputStream>&& output_stream, const arrow::Schema& schema,
+ const std::map<std::string, std::string>& options, const std::string& compression,
+ int32_t batch_size, const std::shared_ptr<MemoryPool>& pool);
+
+ Status AddBatch(ArrowArray* batch) override;
+
+ Status Flush() override;
+
+ Status Finish() override;
+
+ Result<bool> ReachTargetSize(bool suggested_check, int64_t target_size) const override;
+
+ std::shared_ptr<Metrics> GetWriterMetrics() const override;
+
+ private:
+ OrcFormatWriter(const std::shared_ptr<OrcMemoryPool>& orc_memory_pool,
+ std::unique_ptr<::orc::OutputStream>&& output_stream,
+ std::unique_ptr<::orc::WriterMetrics>&& writer_metrics,
+ std::unique_ptr<::orc::Writer>&& writer,
+ std::unique_ptr<::orc::ColumnVectorBatch>&& orc_batch,
+ std::unique_ptr<::orc::Type>&& orc_type,
+ const ::orc::WriterOptions& writer_options,
+ const std::shared_ptr<arrow::DataType>& data_type);
+
+ Result<uint64_t> GetEstimateLength() const;
+ Status ExpandBatch(uint64_t expect_size);
+
+ static Result<::orc::WriterOptions> PrepareWriterOptions(
+ const std::map<std::string, std::string>& options, const std::string& file_compression,
+ const std::shared_ptr<arrow::DataType>& data_type);
+ static Result<::orc::CompressionKind> ToOrcCompressionKind(const std::string& file_compression);
+
+ private:
Review Comment:
`PrepareWriterOptions` and `ToOrcCompressionKind` are declared `private`, but the unit tests call `OrcFormatWriter::PrepareWriterOptions(...)` directly. This will fail to compile unless these helpers are made accessible (or the tests are changed to only validate behavior through the public API).
##########
src/paimon/format/orc/orc_file_batch_reader.h:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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 <list>
+#include <map>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/c/bridge.h"
+#include "arrow/memory_pool.h"
+#include "arrow/type.h"
+#include "orc/OrcFile.hh"
+#include "orc/Reader.hh"
+#include "paimon/format/orc/orc_reader_wrapper.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/predicate/predicate.h"
+#include "paimon/reader/prefetch_file_batch_reader.h"
+
+namespace orc {
+class InputStream;
+} // namespace orc
+
+namespace paimon::orc {
+
+class OrcFileBatchReader : public PrefetchFileBatchReader {
+ public:
+ ~OrcFileBatchReader() override = default;
+ static Result<std::unique_ptr<OrcFileBatchReader>> Create(
+ std::unique_ptr<::orc::InputStream>&& input_stream, const std::shared_ptr<MemoryPool>& pool,
+ const std::map<std::string, std::string>& options, int32_t batch_size);
+
+ // For timestamp type, precision info is missing from file
+ Result<std::unique_ptr<::ArrowSchema>> GetFileSchema() const override;
+
+ Status SetReadSchema(::ArrowSchema* read_schema, const std::shared_ptr<Predicate>& predicate,
+ const std::optional<RoaringBitmap32>& selection_bitmap) override;
+
+ Status SeekToRow(uint64_t row_number) override;
+
+ Status SetReadRanges(const std::vector<std::pair<uint64_t, uint64_t>>& read_ranges) override {
+ return reader_->SetReadRanges(read_ranges);
+ }
+
+ // Important: output ArrowArray is allocated on arrow_pool_ whose lifecycle holds in
+ // OrcFileBatchReader. Therefore, we need to hold BatchReader when using output ArrowArray.
+ Result<ReadBatch> NextBatch() override;
+
+ Result<uint64_t> GetPreviousBatchFirstRowNumber() const override {
+ return reader_->GetRowNumber();
+ }
+
+ Result<uint64_t> GetNumberOfRows() const override {
+ return reader_->GetNumberOfRows();
+ }
+
+ uint64_t GetNextRowToRead() const override {
+ return reader_->GetNextRowToRead();
+ }
Review Comment:
`GetPreviousBatchFirstRowNumber()` is required to return "the row number of the first row in the previously read batch" (see `include/paimon/reader/file_batch_reader.h`). Returning `row_reader_->getRowNumber()`/`GetRowNumber()` is not correct for ORC because `getRowNumber()` reports the *next* row to be read. This will break deletion-vector logic and makes the current tests' expectations (e.g., `-1`/`8`) inconsistent with the contract used by Parquet readers.
##########
src/paimon/format/orc/orc_file_batch_reader_test.cpp:
##########
@@ -0,0 +1,750 @@
+/*
+ * 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.
+ */
+
+#include "paimon/format/orc/orc_file_batch_reader.h"
+
+#include <list>
+#include <map>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/c/bridge.h"
+#include "arrow/ipc/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/defs.h"
+#include "paimon/format/orc/orc_adapter.h"
+#include "paimon/format/orc/orc_format_defs.h"
+#include "paimon/format/orc/orc_format_writer.h"
+#include "paimon/format/orc/orc_input_stream_impl.h"
+#include "paimon/format/orc/orc_memory_pool.h"
+#include "paimon/format/orc/orc_metrics.h"
+#include "paimon/format/orc/orc_output_stream_impl.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/predicate/predicate_builder.h"
+#include "paimon/testing/utils/read_result_collector.h"
+#include "paimon/testing/utils/testharness.h"
+#include "paimon/testing/utils/timezone_guard.h"
+
+namespace paimon::orc::test {
+
+struct TestParam {
+ uint64_t natural_read_size;
+ bool enable_tz;
+};
+
+class OrcFileBatchReaderTest : public ::testing::Test,
+ public ::testing::WithParamInterface<TestParam> {
+ public:
+ void SetUp() override {
+ pool_ = GetDefaultPool();
+ batch_size_ = 10;
+
+ arrow::FieldVector fields = {
+ arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
+ arrow::field("f2", arrow::int32()), arrow::field("f3", arrow::float64())};
+
+ struct_array_ = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+ ["Bob", 10, 0, 12.1], ["Emily", 10, 0, 13.1], ["Tony", 10, 0, 14.1], ["Emily", 10, 0, 15.1],
+ ["Bob", 10, 0, 12.1], ["Alex", 10, 0, 16.1], ["David", 10, 0, 17.1], ["Lily", 10, 0, 17.1]
+ ])")
+ .ValueOrDie());
+ }
+ void TearDown() override {}
+
+ std::pair<std::unique_ptr<OrcFileBatchReader>, std::shared_ptr<arrow::ChunkedArray>>
+ ReadBatchWithCustomizedData(const std::shared_ptr<arrow::Array>& src_array,
+ int32_t write_batch_size, int32_t write_stripe_size,
+ int32_t write_row_index_stride, const arrow::Schema* read_schema,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::optional<RoaringBitmap32>& selection_bitmap,
+ int32_t read_batch_size, double dict_key_size_threshold,
+ bool enable_lazy_decoding) const {
+ arrow::Schema src_schema(src_array->type()->fields());
+ EXPECT_OK_AND_ASSIGN(std::unique_ptr<::orc::Type> orc_type,
+ OrcAdapter::GetOrcType(src_schema));
+ EXPECT_TRUE(orc_type);
+ auto dir = paimon::test::UniqueTestDirectory::Create();
+ EXPECT_TRUE(dir);
+ auto fs = dir->GetFileSystem();
+ std::string data_path = dir->Str() + "/test.data";
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<paimon::OutputStream> output_stream,
+ fs->Create(data_path, /*overwrite=*/true));
+ EXPECT_TRUE(output_stream);
+ EXPECT_OK_AND_ASSIGN(auto orc_output_stream, OrcOutputStreamImpl::Create(output_stream));
+ EXPECT_TRUE(orc_output_stream);
+ ::orc::WriterOptions writer_options;
+ writer_options.setDictionaryKeySizeThreshold(dict_key_size_threshold);
+ if (write_stripe_size != -1) {
+ writer_options.setStripeSize(write_stripe_size);
+ }
+ if (write_row_index_stride != -1) {
+ writer_options.setRowIndexStride(write_row_index_stride);
+ }
+ std::unique_ptr<::orc::Writer> writer =
+ ::orc::createWriter(*orc_type, orc_output_stream.get(), writer_options);
+ // for simple case, assume that src_array.length() % write_batch_size == 0
+ EXPECT_EQ(src_array->length() % write_batch_size, 0);
+ int32_t write_batch_count = src_array->length() / write_batch_size;
+ for (int32_t i = 0; i < write_batch_count; i++) {
+ auto src_slice = src_array->Slice(i * write_batch_size, write_batch_size);
+ auto write_batch = writer->createRowBatch(src_slice->length());
+ // Convert from arrow array to orc batch
+ EXPECT_OK(OrcAdapter::WriteBatch(src_slice, write_batch.get()));
+ writer->add(*write_batch);
+ }
+ writer->close();
+ EXPECT_OK(output_stream->Close());
+
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<paimon::InputStream> input_stream,
+ fs->Open(data_path));
+ EXPECT_TRUE(input_stream);
+ EXPECT_OK_AND_ASSIGN(auto orc_input_stream,
+ OrcInputStreamImpl::Create(input_stream, DEFAULT_NATURAL_READ_SIZE));
+ EXPECT_TRUE(orc_input_stream);
+ std::string enable_lazy_decoding_str = enable_lazy_decoding ? "true" : "false";
+ std::map<std::string, std::string> options = {
+ {ORC_READ_ENABLE_LAZY_DECODING, enable_lazy_decoding_str}};
+ auto orc_batch_reader =
+ PrepareOrcFileBatchReader(std::move(orc_input_stream), options, read_schema, predicate,
+ selection_bitmap, read_batch_size);
+ EXPECT_OK_AND_ASSIGN(
+ auto result, paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ return std::make_pair(std::move(orc_batch_reader), result);
+ }
+
+ std::unique_ptr<OrcFileBatchReader> PrepareOrcFileBatchReader(
+ const std::string& file_name, const arrow::Schema* read_schema, int32_t batch_size,
+ uint64_t natural_read_size) const {
+ std::shared_ptr<FileSystem> file_system = std::make_shared<LocalFileSystem>();
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input_stream,
+ file_system->Open(file_name));
+ EXPECT_TRUE(input_stream);
+ EXPECT_OK_AND_ASSIGN(auto in_stream,
+ OrcInputStreamImpl::Create(input_stream, natural_read_size));
+ EXPECT_TRUE(in_stream);
+ std::map<std::string, std::string> options = {{ORC_READ_ENABLE_LAZY_DECODING, "true"},
+ {ORC_READ_ENABLE_METRICS, "true"},
+ {"orc.timestamp-ltz.legacy.type", "false"}};
+ return PrepareOrcFileBatchReader(std::move(in_stream), options, read_schema,
+ /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt,
+ batch_size);
+ }
+
+ std::unique_ptr<paimon::orc::OrcFileBatchReader> PrepareOrcFileBatchReader(
+ std::unique_ptr<::orc::InputStream>&& in_stream,
+ const std::map<std::string, std::string>& options, const arrow::Schema* read_schema,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::optional<RoaringBitmap32>& selection_bitmap, int32_t batch_size) const {
+ EXPECT_OK_AND_ASSIGN(
+ auto orc_batch_reader,
+ OrcFileBatchReader::Create(std::move(in_stream), pool_, options, batch_size))
+ EXPECT_TRUE(orc_batch_reader);
+ std::unique_ptr<ArrowSchema> c_schema = std::make_unique<ArrowSchema>();
+ auto arrow_status = arrow::ExportSchema(*read_schema, c_schema.get());
+ EXPECT_TRUE(arrow_status.ok());
+ EXPECT_OK(orc_batch_reader->SetReadSchema(c_schema.get(), predicate, selection_bitmap));
+ return orc_batch_reader;
+ }
+
+ void WriteArray(const std::shared_ptr<FileSystem>& fs, const std::string& file_path,
+ const std::shared_ptr<arrow::Array>& src_array,
+ const std::shared_ptr<arrow::Schema>& arrow_schema,
+ const std::map<std::string, std::string>& options) const {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> out,
+ fs->Create(file_path, /*overwrite=*/true));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OrcOutputStreamImpl> output_stream,
+ OrcOutputStreamImpl::Create(out));
+ ASSERT_OK_AND_ASSIGN(
+ auto format_writer,
+ OrcFormatWriter::Create(std::move(output_stream), *arrow_schema, options, "zstd",
+ /*batch_size=*/src_array->length(), pool_));
+ auto arrow_array = std::make_unique<ArrowArray>();
+ ASSERT_TRUE(arrow::ExportArray(*src_array, arrow_array.get()).ok());
+ ASSERT_OK(format_writer->AddBatch(arrow_array.get()));
+ ASSERT_OK(format_writer->Flush());
+ ASSERT_OK(format_writer->Finish());
+ ASSERT_OK(out->Flush());
+ ASSERT_OK(out->Close());
+ }
+
+ private:
+ std::shared_ptr<MemoryPool> pool_;
+ int32_t batch_size_;
+ std::shared_ptr<arrow::StructArray> struct_array_;
+};
+
+INSTANTIATE_TEST_SUITE_P(TestParam, OrcFileBatchReaderTest,
+ ::testing::Values(TestParam{128 * 1024, false}, TestParam{16, false},
+ TestParam{16, true}));
+
+TEST_F(OrcFileBatchReaderTest, TestSetReadSchema) {
+ std::string file_name = paimon::test::GetDataDir() +
+ "/orc/append_09.db/append_09/f1=10/bucket-1/"
+ "data-b9e7c41f-66e8-4dad-b25a-e6e1963becc4-0.orc";
+ std::shared_ptr<FileSystem> file_system = std::make_shared<LocalFileSystem>();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input_stream, file_system->Open(file_name));
+ ASSERT_OK_AND_ASSIGN(auto in_stream,
+ OrcInputStreamImpl::Create(input_stream, DEFAULT_NATURAL_READ_SIZE));
+ std::map<std::string, std::string> options = {{ORC_READ_ENABLE_LAZY_DECODING, "true"}};
+ ASSERT_OK_AND_ASSIGN(
+ auto orc_batch_reader,
+ OrcFileBatchReader::Create(std::move(in_stream), pool_, options, batch_size_));
+ // test GetFileSchema()
+ ASSERT_OK_AND_ASSIGN(auto c_file_schema, orc_batch_reader->GetFileSchema());
+ auto arrow_file_schema = arrow::ImportSchema(c_file_schema.get()).ValueOrDie();
+ arrow::FieldVector fields = {
+ arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
+ arrow::field("f2", arrow::int32()), arrow::field("f3", arrow::float64())};
+ arrow::Schema file_schema(fields);
+ ASSERT_TRUE(arrow_file_schema->Equals(file_schema));
+ // NextBatch() without SetReadSchema(), will return data with
+ // file schema
+ ASSERT_OK_AND_ASSIGN(auto result_with_file_schema,
+ paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ auto expected_file_chunk_array = std::make_shared<arrow::ChunkedArray>(struct_array_);
+ ASSERT_TRUE(result_with_file_schema->Equals(expected_file_chunk_array));
+ // NextBatch() with SetReadSchema(), will return data with read schema
+ arrow::Schema read_schema({fields[0], fields[2]});
+ std::unique_ptr<ArrowSchema> c_read_schema = std::make_unique<ArrowSchema>();
+ ASSERT_TRUE(arrow::ExportSchema(read_schema, c_read_schema.get()).ok());
+ ASSERT_OK(orc_batch_reader->SetReadSchema(c_read_schema.get(),
+ /*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt));
+ auto expected_read_array =
+ arrow::StructArray::Make({struct_array_->field(0), struct_array_->field(2)},
+ {fields[0], fields[2]})
+ .ValueOrDie();
+ auto expected_read_chunk_array = std::make_shared<arrow::ChunkedArray>(expected_read_array);
+ ASSERT_OK_AND_ASSIGN(auto result_with_read_schema,
+ paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ ASSERT_TRUE(result_with_read_schema->Equals(expected_read_chunk_array));
+
+ // NextBatch() with predicate
+ auto predicate = PredicateBuilder::GreaterThan(
+ /*field_index=*/3, /*field_name=*/"f2", FieldType::INT, Literal(2));
+ c_read_schema = std::make_unique<ArrowSchema>();
+ ASSERT_TRUE(arrow::ExportSchema(read_schema, c_read_schema.get()).ok());
+ ASSERT_OK(orc_batch_reader->SetReadSchema(c_read_schema.get(), predicate,
+ /*selection_bitmap=*/std::nullopt));
+ ASSERT_OK_AND_ASSIGN(result_with_read_schema,
+ paimon::test::ReadResultCollector::CollectResult(orc_batch_reader.get()));
+ ASSERT_FALSE(result_with_read_schema);
+}
+
+TEST_F(OrcFileBatchReaderTest, TestCreateRowReaderOptions) {
+ auto orc_pool = std::make_shared<OrcMemoryPool>(pool_);
+ std::vector<uint64_t> target_column_ids;
+ {
+ // read all fields && default options
+ std::map<std::string, std::string> options;
+ std::string orc_schema = "struct<col1:int,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(orc_schema);
+ std::unique_ptr<::orc::Type> target_type = ::orc::Type::buildTypeFromString(orc_schema);
+ ASSERT_OK_AND_ASSIGN(auto row_reader_option,
+ OrcFileBatchReader::CreateRowReaderOptions(
+ src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options, &target_column_ids));
+ ASSERT_EQ(std::list<std::string>({"col1", "col2", "col3"}),
+ row_reader_option.getIncludeNames());
+ ASSERT_EQ(row_reader_option.getEnableLazyDecoding(), false);
+ }
+ {
+ // read partial fields && set options
+ std::map<std::string, std::string> options = {{ORC_READ_ENABLE_LAZY_DECODING, "true"}};
+ std::string src_orc_schema = "struct<col1:int,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(src_orc_schema);
+ std::string target_orc_schema = "struct<col1:int,col3:string>";
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString(target_orc_schema);
+ ASSERT_OK_AND_ASSIGN(auto row_reader_option,
+ OrcFileBatchReader::CreateRowReaderOptions(
+ src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options, &target_column_ids));
+ ASSERT_EQ(std::list<std::string>({"col1", "col3"}), row_reader_option.getIncludeNames());
+ ASSERT_EQ(row_reader_option.getEnableLazyDecoding(), true);
+ }
+ {
+ // read partial fields, sequence mismatch
+ std::map<std::string, std::string> options;
+ std::string src_orc_schema = "struct<col1:int,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(src_orc_schema);
+ std::string target_orc_schema = "struct<col3:string,col1:int>";
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString(target_orc_schema);
+ ASSERT_NOK_WITH_MSG(
+ OrcFileBatchReader::CreateRowReaderOptions(src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options,
+ &target_column_ids),
+ "The column id of the target field should be monotonically increasing in format "
+ "reader");
+ }
+ {
+ // read non exist column
+ std::map<std::string, std::string> options;
+ std::string src_orc_schema = "struct<col1:int,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(src_orc_schema);
+ std::string target_orc_schema = "struct<col1:int,non_exist_col:double>";
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString(target_orc_schema);
+ ASSERT_NOK_WITH_MSG(OrcFileBatchReader::CreateRowReaderOptions(
+ src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options, &target_column_ids),
+ "field non_exist_col not in file schema");
+ }
+ {
+ std::map<std::string, std::string> options;
+ std::string src_orc_schema =
+ "struct<col1:struct<sub1:int,sub2:array<array<double>>,sub3:int>,col2:double,col3:"
+ "map<string,int>>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(src_orc_schema);
+ std::string target_orc_schema =
+ "struct<col1:struct<sub1:int,sub2:array<array<double>>,sub3:int>,col3:map<string,int>>";
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString(target_orc_schema);
+ ASSERT_OK_AND_ASSIGN(auto row_reader_option,
+ OrcFileBatchReader::CreateRowReaderOptions(
+ src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options, &target_column_ids));
+ ASSERT_EQ(std::list<std::string>({"col1", "col3"}), row_reader_option.getIncludeNames());
+ }
+ {
+ // read with type mismatch
+ std::map<std::string, std::string> options;
+ std::string src_orc_schema =
+ "struct<col1:struct<sub1:int,sub2:int,sub3:int>,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(src_orc_schema);
+ std::string target_orc_schema =
+ "struct<col1:struct<sub1:int,sub2:double,sub3:int>,col2:double,col3:string>";
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString(target_orc_schema);
+
+ ASSERT_NOK_WITH_MSG(
+ OrcFileBatchReader::CreateRowReaderOptions(src_type.get(), target_type.get(),
+ /*search_arg=*/nullptr, options,
+ &target_column_ids),
+ "target_type "
+ "struct<col1:struct<sub1:int,sub2:double,sub3:int>,col2:double,col3:string> not match "
+ "src_type struct<col1:struct<sub1:int,sub2:int,sub3:int>,col2:double,col3:string>, "
+ "mismatch field name col1");
+ }
+}
+
+TEST_P(OrcFileBatchReaderTest, TestNextBatchSimple) {
+ std::string file_name = paimon::test::GetDataDir() +
+ "/orc/append_09.db/append_09/f1=10/bucket-1/"
+ "data-b9e7c41f-66e8-4dad-b25a-e6e1963becc4-0.orc";
+ auto data_type = struct_array_->struct_type();
+ arrow::Schema read_schema(data_type->fields());
+ auto [natural_read_size, _] = GetParam();
+ for (auto batch_size : {1, 2, 3, 5, 8, 10}) {
+ auto orc_batch_reader =
+ PrepareOrcFileBatchReader(file_name, &read_schema, batch_size, natural_read_size);
+ ASSERT_EQ(orc_batch_reader->GetPreviousBatchFirstRowNumber().value(), -1);
+ ASSERT_OK_AND_ASSIGN(auto result_array, paimon::test::ReadResultCollector::CollectResult(
+ orc_batch_reader.get()));
+ ASSERT_EQ(orc_batch_reader->GetPreviousBatchFirstRowNumber().value(), 8);
+ orc_batch_reader->Close();
Review Comment:
This test asserts `GetPreviousBatchFirstRowNumber()` equals `-1` initially and `8` after reading, but the API contract is "first row of the previously read batch" (see `include/paimon/reader/file_batch_reader.h`). Parquet tests use `std::numeric_limits<uint64_t>::max()` as the initial sentinel and then assert the *start row* of each batch; this test should be updated accordingly once ORC tracks `previous_batch_first_row_number_` correctly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: add ORC format reader/writer utilities [paimon-cpp]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #122:
URL: https://github.com/apache/paimon-cpp/pull/122
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org