zhouyc-ali commented on code in PR #471:
URL: https://github.com/apache/paimon-cpp/pull/471#discussion_r4477283316
##########
src/paimon/format/vortex/vortex_file_batch_reader.h:
##########
@@ -96,30 +155,26 @@ class VortexFileBatchReader : public FileBatchReader {
VxSessionPtr session_;
VxDataSourcePtr data_source_;
VxScanPtr scan_;
- // The file schema with Vortex's string/binary view types normalized to
the standard types; this
- // is what GetFileSchema() reports to paimon.
+ // Full file schema with standard Arrow types, as reported by
GetFileSchema().
std::shared_ptr<arrow::Schema> file_schema_;
- // The RAW view-typed struct that Vortex's Arrow export produces.
NextBatch imports each scanned
- // batch as this type (so it must match the C export's buffer layout),
then NormalizeViewArray
- // rewrites it to file_schema_'s types. Keeping this view-typed is
load-bearing: normalizing it
- // would make ImportArray misread the buffers.
- std::shared_ptr<arrow::DataType> struct_type_;
// The schema NextBatch must output, set by SetReadSchema (the projected
read schema). Null
// until SetReadSchema is called, in which case the full file schema is
returned.
std::shared_ptr<arrow::Schema> read_schema_;
uint64_t total_rows_ = 0;
::ArrowArrayStream current_stream_{};
+ // Actual scan output schema, including projection and any hidden physical
row index.
+ std::shared_ptr<arrow::Schema> current_stream_schema_;
bool stream_active_ = false;
std::shared_ptr<arrow::Array> current_batch_;
int64_t current_batch_offset_ = 0;
- uint64_t rows_emitted_ = 0;
- uint64_t previous_first_row_ = std::numeric_limits<uint64_t>::max();
- uint64_t previous_batch_row_count_ = 0;
+ SelectionState selection_;
+ RowPositionState row_position_;
Review Comment:
Reset both states in `CloseInternal()`, releasing vector capacity and the
row-index array. Added cleanup and output batch lifetime tests.
##########
src/paimon/format/vortex/vortex_file_batch_reader.cpp:
##########
@@ -232,26 +149,152 @@ std::shared_ptr<arrow::ArrayData>
AlignTimestampTimezones(
} // namespace
+Result<VortexFileBatchReader::SelectionState>
VortexFileBatchReader::SelectionState::Create(
+ const std::optional<RoaringBitmap32>& bitmap, uint64_t total_rows) {
+ SelectionState state;
+ state.has_selection_ = bitmap.has_value();
+ state.selected_row_count_ = total_rows;
+ if (bitmap) {
+ const int32_t cardinality = bitmap->Cardinality();
+ if (cardinality < 0 || static_cast<uint64_t>(cardinality) >
total_rows) {
+ return Status::Invalid("Vortex selection cardinality exceeds the
file row count");
+ }
+ state.selected_row_count_ = static_cast<uint64_t>(cardinality);
+ state.excludes_ = state.selected_row_count_ > total_rows / 2;
+ state.indices_.reserve(static_cast<size_t>(
+ state.excludes_ ? total_rows - state.selected_row_count_ :
state.selected_row_count_));
+ uint64_t next_row = 0;
+ const auto end = bitmap->End();
+ for (auto iter = bitmap->Begin(); iter != end; ++iter) {
Review Comment:
Bounds are checked first; full selections return early and dense selections
enumerate the compressed complement. Added boundary tests and benchmarked large
full selections and single deletions.
##########
src/paimon/format/vortex/vortex_file_batch_reader.cpp:
##########
@@ -306,28 +347,67 @@ Result<std::unique_ptr<VortexFileBatchReader>>
VortexFileBatchReader::Create(
// scan may be consumed only once.
return std::unique_ptr<VortexFileBatchReader>(new VortexFileBatchReader(
input, batch_size, std::move(input_context), std::move(session),
std::move(data_source),
- VxScanPtr(nullptr, vx_scan_free), file_schema, struct_type,
total_rows, arrow_pool));
+ VxScanPtr(nullptr, vx_scan_free), file_schema, total_rows,
predicate_options, arrow_pool));
}
VortexFileBatchReader::~VortexFileBatchReader() {
CloseInternal();
}
+Status VortexFileBatchReader::CreateScan() {
+ vx_error* error = nullptr;
+ // Scan in storage order: Vortex defaults to ordered=false and may emit
chunks out of order,
+ // which would misalign the physical row positions NextBatch assigns.
+ vx_scan_options options{};
+ options.ordered = true;
+ // The scan clones the projection, so these expressions need only outlive
scan creation.
+ VxExpressionPtr root(nullptr, vx_expression_free);
+ VxExpressionPtr projection(nullptr, vx_expression_free);
+ std::vector<const char*> name_ptrs;
+ if (read_schema_ != nullptr && read_schema_->num_fields() > 0) {
Review Comment:
Explicitly empty schemas now produce an empty native projection. Added
native schema and I/O assertions verifying omitted payload columns are skipped.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
