EIC code displayed by LXR master/​include/​parquet/​stream_reader.h	Source navigation Diff markup Identifier search General search
	Version: master test Revert   Change

File indexing completed on 2026-04-17 08:28:55

 // 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 <array>
 #include <chrono>
 #include <cstdint>
 #include <cstring>
 #include <memory>
 #include <optional>
 #include <string>
 #include <vector>
 
 #include "parquet/column_reader.h"
 #include "parquet/file_reader.h"
 #include "parquet/stream_writer.h"
 
 namespace parquet {
 
 /// \brief A class for reading Parquet files using an output stream type API.
 ///
 /// The values given must be of the correct type i.e. the type must
 /// match the file schema exactly otherwise a ParquetException will be
 /// thrown.
 ///
 /// The user must explicitly advance to the next row using the
 /// EndRow() function or EndRow input manipulator.
 ///
 /// Required and optional fields are supported:
 /// - Required fields are read using operator>>(T)
 /// - Optional fields are read with
 ///   operator>>(std::optional<T>)
 ///
 /// Note that operator>>(std::optional<T>) can be used to read
 /// required fields.
 ///
 /// Similarly operator>>(T) can be used to read optional fields.
 /// However, if the value is not present then a ParquetException will
 /// be raised.
 ///
 /// Currently there is no support for repeated fields.
 ///
 class PARQUET_EXPORT StreamReader {
  public:
   template <typename T>
   using optional = ::std::optional<T>;
 
   // N.B. Default constructed objects are not usable.  This
   //      constructor is provided so that the object may be move
   //      assigned afterwards.
   StreamReader() = default;
 
   explicit StreamReader(std::unique_ptr<ParquetFileReader> reader);
 
   ~StreamReader() = default;
 
   bool eof() const { return eof_; }
 
   int current_column() const { return column_index_; }
 
   int64_t current_row() const { return current_row_; }
 
   int num_columns() const;
 
   int64_t num_rows() const;
 
   // Moving is possible.
   StreamReader(StreamReader&&) = default;
   StreamReader& operator=(StreamReader&&) = default;
 
   // Copying is not allowed.
   StreamReader(const StreamReader&) = delete;
   StreamReader& operator=(const StreamReader&) = delete;
 
   StreamReader& operator>>(bool& v);
 
   StreamReader& operator>>(int8_t& v);
 
   StreamReader& operator>>(uint8_t& v);
 
   StreamReader& operator>>(int16_t& v);
 
   StreamReader& operator>>(uint16_t& v);
 
   StreamReader& operator>>(int32_t& v);
 
   StreamReader& operator>>(uint32_t& v);
 
   StreamReader& operator>>(int64_t& v);
 
   StreamReader& operator>>(uint64_t& v);
 
   StreamReader& operator>>(std::chrono::milliseconds& v);
 
   StreamReader& operator>>(std::chrono::microseconds& v);
 
   StreamReader& operator>>(float& v);
 
   StreamReader& operator>>(double& v);
 
   StreamReader& operator>>(char& v);
 
   template <int N>
   StreamReader& operator>>(char (&v)[N]) {
     ReadFixedLength(v, N);
     return *this;
   }
 
   template <std::size_t N>
   StreamReader& operator>>(std::array<char, N>& v) {
     ReadFixedLength(v.data(), static_cast<int>(N));
     return *this;
   }
 
   // N.B. Cannot allow for reading to a arbitrary char pointer as the
   //      length cannot be verified.  Also it would overshadow the
   //      char[N] input operator.
   // StreamReader& operator>>(char * v);
 
   StreamReader& operator>>(std::string& v);
 
   StreamReader& operator>>(::arrow::Decimal128& v);
 
   // Input operators for optional fields.
 
   StreamReader& operator>>(optional<bool>& v);
 
   StreamReader& operator>>(optional<int8_t>& v);
 
   StreamReader& operator>>(optional<uint8_t>& v);
 
   StreamReader& operator>>(optional<int16_t>& v);
 
   StreamReader& operator>>(optional<uint16_t>& v);
 
   StreamReader& operator>>(optional<int32_t>& v);
 
   StreamReader& operator>>(optional<uint32_t>& v);
 
   StreamReader& operator>>(optional<int64_t>& v);
 
   StreamReader& operator>>(optional<uint64_t>& v);
 
   StreamReader& operator>>(optional<float>& v);
 
   StreamReader& operator>>(optional<double>& v);
 
   StreamReader& operator>>(optional<std::chrono::milliseconds>& v);
 
   StreamReader& operator>>(optional<std::chrono::microseconds>& v);
 
   StreamReader& operator>>(optional<char>& v);
 
   StreamReader& operator>>(optional<std::string>& v);
 
   StreamReader& operator>>(optional<::arrow::Decimal128>& v);
 
   template <std::size_t N>
   StreamReader& operator>>(optional<std::array<char, N>>& v) {
     CheckColumn(Type::FIXED_LEN_BYTE_ARRAY, ConvertedType::NONE, N);
     FixedLenByteArray flba;
     if (ReadOptional(&flba)) {
       v = std::array<char, N>{};
       std::memcpy(v->data(), flba.ptr, N);
     } else {
       v.reset();
     }
     return *this;
   }
 
   /// \brief Terminate current row and advance to next one.
   /// \throws ParquetException if all columns in the row were not
   /// read or skipped.
   void EndRow();
 
   /// \brief Skip the data in the next columns.
   /// If the number of columns exceeds the columns remaining on the
   /// current row then skipping is terminated - it does _not_ continue
   /// skipping columns on the next row.
   /// Skipping of columns still requires the use 'EndRow' even if all
   /// remaining columns were skipped.
   /// \return Number of columns actually skipped.
   int64_t SkipColumns(int64_t num_columns_to_skip);
 
   /// \brief Skip the data in the next rows.
   /// Skipping of rows is not allowed if reading of data for the
   /// current row is not finished.
   /// Skipping of rows will be terminated if the end of file is
   /// reached.
   /// \return Number of rows actually skipped.
   int64_t SkipRows(int64_t num_rows_to_skip);
 
  protected:
   [[noreturn]] void ThrowReadFailedException(
       const std::shared_ptr<schema::PrimitiveNode>& node);
 
   template <typename ReaderType, typename T>
   void Read(T* v) {
     const auto& node = nodes_[column_index_];
     auto reader = static_cast<ReaderType*>(column_readers_[column_index_++].get());
     int16_t def_level;
     int16_t rep_level;
     int64_t values_read;
 
     reader->ReadBatch(kBatchSizeOne, &def_level, &rep_level, v, &values_read);
 
     if (values_read != 1) {
       ThrowReadFailedException(node);
     }
   }
 
   template <typename ReaderType, typename ReadType, typename T>
   void Read(T* v) {
     const auto& node = nodes_[column_index_];
     auto reader = static_cast<ReaderType*>(column_readers_[column_index_++].get());
     int16_t def_level;
     int16_t rep_level;
     ReadType tmp;
     int64_t values_read;
 
     reader->ReadBatch(kBatchSizeOne, &def_level, &rep_level, &tmp, &values_read);
 
     if (values_read == 1) {
       *v = tmp;
     } else {
       ThrowReadFailedException(node);
     }
   }
 
   template <typename ReaderType, typename ReadType = typename ReaderType::T, typename T>
   void ReadOptional(optional<T>* v) {
     const auto& node = nodes_[column_index_];
     auto reader = static_cast<ReaderType*>(column_readers_[column_index_++].get());
     int16_t def_level;
     int16_t rep_level;
     ReadType tmp;
     int64_t values_read;
 
     reader->ReadBatch(kBatchSizeOne, &def_level, &rep_level, &tmp, &values_read);
 
     if (values_read == 1) {
       *v = T(tmp);
     } else if ((values_read == 0) && (def_level == 0)) {
       v->reset();
     } else {
       ThrowReadFailedException(node);
     }
   }
 
   void ReadFixedLength(char* ptr, int len);
 
   void Read(ByteArray* v);
 
   void Read(FixedLenByteArray* v);
 
   bool ReadOptional(ByteArray* v);
 
   bool ReadOptional(FixedLenByteArray* v);
 
   void NextRowGroup();
 
   void CheckColumn(Type::type physical_type, ConvertedType::type converted_type,
                    int length = 0);
 
   void SkipRowsInColumn(ColumnReader* reader, int64_t num_rows_to_skip);
 
   void SetEof();
 
  private:
   std::unique_ptr<ParquetFileReader> file_reader_;
   std::shared_ptr<FileMetaData> file_metadata_;
   std::shared_ptr<RowGroupReader> row_group_reader_;
   std::vector<std::shared_ptr<ColumnReader>> column_readers_;
   std::vector<std::shared_ptr<schema::PrimitiveNode>> nodes_;
 
   bool eof_{true};
   int row_group_index_{0};
   int column_index_{0};
   int64_t current_row_{0};
   int64_t row_group_row_offset_{0};
 
   static constexpr int64_t kBatchSizeOne = 1;
 };  // namespace parquet
 
 PARQUET_EXPORT
 StreamReader& operator>>(StreamReader&, EndRowType);
 
 }  // namespace parquet

This page was automatically generated by the 2.3.7 LXR engine. The LXR team