diff --git a/cpp/src/arrow/compute/kernels/scalar_cast_internal.cc b/cpp/src/arrow/compute/kernels/scalar_cast_internal.cc
index 0484e0b259c..a56f2ed1298 100644
--- a/cpp/src/arrow/compute/kernels/scalar_cast_internal.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_cast_internal.cc
@@ -209,6 +209,13 @@ Status UnpackDictionary(KernelContext* ctx, const ExecSpan& batch, ExecResult* o
   return Status::OK();
 }
 
+Status DecodeRunEndEncoded(KernelContext* ctx, const ExecSpan& batch, ExecResult* out) {
+  Datum input(batch[0].array.ToArrayData());
+  ARROW_ASSIGN_OR_RAISE(Datum decoded, RunEndDecode(input, ctx->exec_context()));
+  out->value = std::move(decoded.array());
+  return Status::OK();
+}
+
 Status OutputAllNull(KernelContext* ctx, const ExecSpan& batch, ExecResult* out) {
   // TODO(wesm): there is no good reason to have to use ArrayData here, so we
   // should clean this up later. This is used in the dict<null>->null cast
@@ -286,6 +293,12 @@ static bool CanCastFromDictionary(Type::type type_id) {
           is_fixed_size_binary(type_id));
 }
 
+static bool CanCastFromRunEndEncoded(Type::type type_id) {
+  return (is_primitive(type_id) ||
+          is_base_binary_like(type_id) ||
+          is_fixed_size_binary(type_id));
+}
+
 void AddCommonCasts(Type::type out_type_id, OutputType out_ty, CastFunction* func) {
   // From null to this type
   ScalarKernel kernel;
@@ -303,6 +316,13 @@ void AddCommonCasts(Type::type out_type_id, OutputType out_ty, CastFunction* fun
                               MemAllocation::NO_PREALLOCATE));
   }
 
+  // From run-end-encoded to this type
+  if (CanCastFromRunEndEncoded(out_type_id)) {
+    DCHECK_OK(func->AddKernel(Type::RUN_END_ENCODED, {InputType(Type::RUN_END_ENCODED)}, out_ty,
+                              DecodeRunEndEncoded, NullHandling::COMPUTED_NO_PREALLOCATE,
+                              MemAllocation::NO_PREALLOCATE));
+  }
+
   // From extension type to this type
   DCHECK_OK(func->AddKernel(Type::EXTENSION, {InputType(Type::EXTENSION)}, out_ty,
                             CastFromExtension, NullHandling::COMPUTED_NO_PREALLOCATE,
diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
index 7ef60618700..f62d52cc7d9 100644
--- a/cpp/src/arrow/dataset/file_parquet.cc
+++ b/cpp/src/arrow/dataset/file_parquet.cc
@@ -115,6 +115,10 @@ parquet::ArrowReaderProperties MakeArrowReaderProperties(
     auto column_index = metadata.schema()->ColumnIndex(name);
     properties.set_read_dictionary(column_index, true);
   }
+  for (const std::string& name : format.reader_options.ree_columns) {
+    auto column_index = metadata.schema()->ColumnIndex(name);
+    properties.set_read_ree(column_index, true);
+  }
   properties.set_coerce_int96_timestamp_unit(
       format.reader_options.coerce_int96_timestamp_unit);
   properties.set_binary_type(format.reader_options.binary_type);
@@ -445,6 +449,7 @@ bool ParquetFileFormat::Equals(const FileFormat& other) const {
 
   // FIXME implement comparison for decryption options
   return (reader_options.dict_columns == other_reader_options.dict_columns &&
+          reader_options.ree_columns == other_reader_options.ree_columns &&
           reader_options.coerce_int96_timestamp_unit ==
               other_reader_options.coerce_int96_timestamp_unit &&
           reader_options.binary_type == other_reader_options.binary_type &&
diff --git a/cpp/src/arrow/dataset/file_parquet.h b/cpp/src/arrow/dataset/file_parquet.h
index 1811a96bf98..05126e334f8 100644
--- a/cpp/src/arrow/dataset/file_parquet.h
+++ b/cpp/src/arrow/dataset/file_parquet.h
@@ -89,6 +89,7 @@ class ARROW_DS_EXPORT ParquetFileFormat : public FileFormat {
     ///
     /// @{
     std::unordered_set<std::string> dict_columns;
+    std::unordered_set<std::string> ree_columns;
     arrow::TimeUnit::type coerce_int96_timestamp_unit = arrow::TimeUnit::NANO;
     Type::type binary_type = Type::BINARY;
     Type::type list_type = Type::LIST;
diff --git a/cpp/src/arrow/dataset/file_parquet_test.cc b/cpp/src/arrow/dataset/file_parquet_test.cc
index 696bda19359..4d0771a775a 100644
--- a/cpp/src/arrow/dataset/file_parquet_test.cc
+++ b/cpp/src/arrow/dataset/file_parquet_test.cc
@@ -271,6 +271,17 @@ TEST_F(TestParquetFileFormat, InspectDictEncoded) {
   AssertSchemaEqual(*actual, expected_schema, /* check_metadata = */ false);
 }
 
+TEST_F(TestParquetFileFormat, InspectReeEncoded) {
+  auto reader = GetRecordBatchReader(schema({field("utf8", utf8())}));
+  auto source = GetFileSource(reader.get());
+
+  format_->reader_options.ree_columns = {"utf8"};
+  ASSERT_OK_AND_ASSIGN(auto actual, format_->Inspect(*source.get()));
+
+  Schema expected_schema({field("utf8", run_end_encoded(int32(), utf8()))});
+  AssertSchemaEqual(*actual, expected_schema, /* check_metadata = */ false);
+}
+
 TEST_F(TestParquetFileFormat, IsSupported) { TestIsSupported(); }
 
 TEST_F(TestParquetFileFormat, WriteRecordBatchReader) { TestWrite(); }
@@ -617,6 +628,25 @@ TEST_P(TestParquetFileFormatScan, ScanRecordBatchReaderDictEncoded) {
   }
   ASSERT_EQ(row_count, expected_rows());
 }
+TEST_P(TestParquetFileFormatScan, ScanRecordBatchReaderReeEncoded) {
+  auto reader = GetRecordBatchReader(schema({field("utf8", utf8())}));
+  auto source = GetFileSource(reader.get());
+
+  SetSchema(reader->schema()->fields());
+  SetFilter(literal(true));
+  format_->reader_options.ree_columns = {"utf8"};
+  ASSERT_OK_AND_ASSIGN(auto fragment, format_->MakeFragment(*source));
+
+  int64_t row_count = 0;
+  Schema expected_schema({field("utf8", run_end_encoded(int32(), utf8()))});
+
+  for (auto maybe_batch : PhysicalBatches(fragment)) {
+    ASSERT_OK_AND_ASSIGN(auto batch, maybe_batch);
+    row_count += batch->num_rows();
+    AssertSchemaEqual(*batch->schema(), expected_schema, /* check_metadata = */ false);
+  }
+  ASSERT_EQ(row_count, expected_rows());
+}
 TEST_P(TestParquetFileFormatScan, ScanRecordBatchReaderPreBuffer) {
   auto reader = GetRecordBatchReader(schema({field("f64", float64())}));
   auto source = GetFileSource(reader.get());
diff --git a/cpp/src/arrow/util/rle_encoding_internal.h b/cpp/src/arrow/util/rle_encoding_internal.h
index 2420270f3ab..b330c14f0bc 100644
--- a/cpp/src/arrow/util/rle_encoding_internal.h
+++ b/cpp/src/arrow/util/rle_encoding_internal.h
@@ -426,6 +426,14 @@ class RleBitPackedDecoder {
   /// values.
   [[nodiscard]] bool Get(value_type* val);
 
+  // Get the next logical value and num_repeats within the specified batch_size.
+  [[nodiscard]] bool GetNextValueAndNumRepeats(value_type* val, int* num_repeats, int batch_size);
+
+  /// Like GetNextValueAndNumRepeats but add spacing for null entries.
+  [[nodiscard]] bool GetNextValueAndNumRepeatsSpaced(value_type* val, bool* is_null,
+                                             int* num_repeats, int batch_size,
+                                             const uint8_t* valid_bits, int64_t valid_bits_offset);
+
   /// Get a batch of values return the number of decoded elements.
   /// May write fewer elements to the output than requested if there are not enough values
   /// left or if an error occurred.
@@ -722,7 +730,6 @@ void RleBitPackedDecoder<T>::ParseWithCallable(Callable&& func) {
     auto OnBitPackedRun(BitPackedRun run) { return func(std::move(run)); }
     auto OnRleRun(RleRun run) { return func(std::move(run)); }
   } handler{std::move(func)};
-
   parser_.Parse(std::move(handler));
 }
 
@@ -731,6 +738,69 @@ bool RleBitPackedDecoder<T>::Get(value_type* val) {
   return GetBatch(val, 1) == 1;
 }
 
+template <typename T>
+bool RleBitPackedDecoder<T>::GetNextValueAndNumRepeats(value_type* val, int* num_repeats, int batch_size) {
+  using ControlFlow = RleBitPackedParser::ControlFlow;
+
+  if (ARROW_PREDICT_FALSE(run_remaining() > 0)) {
+    if (std::holds_alternative<BitPackedRunDecoder<value_type>>(decoder_)) {
+      auto& decoder = std::get<BitPackedRunDecoder<value_type>>(decoder_);
+      *num_repeats = 1;
+      return decoder.Get(val, value_bit_width_);
+    } else {
+      auto& decoder = std::get<RleRunDecoder<value_type>>(decoder_);
+      *num_repeats = std::min(decoder.remaining(), batch_size);
+      ARROW_DCHECK_EQ(decoder.Advance(*num_repeats, value_bit_width_), *num_repeats);
+      return decoder.Get(val, value_bit_width_);
+    }
+  }
+
+  bool read_new_value = false;
+
+  ParseWithCallable([&](auto run) {
+    if constexpr(std::is_same_v<decltype(run), BitPackedRun>) {
+      BitPackedRunDecoder<T> decoder(run, value_bit_width_);
+      read_new_value = decoder.Get(val, value_bit_width_);
+      *num_repeats = 1;
+      decoder_ = std::move(decoder);
+      return ControlFlow::Break;
+    }
+    else {
+      RleRunDecoder<T> decoder(run, value_bit_width_);
+      *num_repeats = std::min(decoder.remaining(), batch_size);
+      read_new_value = decoder.Get(val, value_bit_width_);
+      ARROW_DCHECK_EQ(decoder.Advance(*num_repeats, value_bit_width_), *num_repeats);
+      decoder_ = std::move(decoder);
+      return ControlFlow::Break;
+    }
+  });
+
+  return read_new_value;
+}
+
+template <typename T>
+bool RleBitPackedDecoder<T>::GetNextValueAndNumRepeatsSpaced(value_type* val, bool* is_null,
+                                                     int* num_repeats, int batch_size,
+                                                     const uint8_t* valid_bits, int64_t valid_bits_offset) {
+  arrow::internal::BitRunReader bit_reader(valid_bits, valid_bits_offset,
+                                           /*length=*/batch_size);
+  arrow::internal::BitRun valid_run = bit_reader.NextRun();
+  while (ARROW_PREDICT_FALSE(valid_run.length == 0)) {
+    valid_run = bit_reader.NextRun();
+  }
+  ARROW_DCHECK_GT(batch_size, 0);
+  ARROW_DCHECK_GT(valid_run.length, 0);
+  if (valid_run.set) {
+    return GetNextValueAndNumRepeats(
+        val, num_repeats,
+        static_cast<int>(std::min(valid_run.length, static_cast<int64_t>(batch_size))));
+  } else {
+    *is_null = true;
+    *num_repeats = static_cast<int>(valid_run.length);
+  }
+  return true;
+}
+
 template <typename T>
 auto RleBitPackedDecoder<T>::GetBatch(value_type* out, rle_size_t batch_size)
     -> rle_size_t {
diff --git a/cpp/src/parquet/CMakeLists.txt b/cpp/src/parquet/CMakeLists.txt
index dc7d40d2a38..a88aa7f1452 100644
--- a/cpp/src/parquet/CMakeLists.txt
+++ b/cpp/src/parquet/CMakeLists.txt
@@ -406,7 +406,10 @@ add_parquet_test(arrow-reader-writer-test
                  SOURCES
                  arrow/arrow_reader_writer_test.cc
                  arrow/arrow_statistics_test.cc
-                 arrow/variant_test.cc)
+                 arrow/variant_test.cc
+                 EXTRA_LINK_LIBS
+                 arrow_compute_core_testing
+                 arrow_acero_testing)
 
 add_parquet_test(arrow-internals-test SOURCES arrow/path_internal_test.cc
                  arrow/reconstruct_internal_test.cc)
diff --git a/cpp/src/parquet/arrow/arrow_reader_writer_test.cc b/cpp/src/parquet/arrow/arrow_reader_writer_test.cc
index e081b428e24..d691393c17f 100644
--- a/cpp/src/parquet/arrow/arrow_reader_writer_test.cc
+++ b/cpp/src/parquet/arrow/arrow_reader_writer_test.cc
@@ -37,6 +37,7 @@
 #include "arrow/array/builder_primitive.h"
 #include "arrow/chunked_array.h"
 #include "arrow/compute/api.h"
+#include "arrow/compute/initialize.h"
 #include "arrow/extension/json.h"
 #include "arrow/io/api.h"
 #include "arrow/record_batch.h"
@@ -61,6 +62,8 @@
 #  include "arrow/csv/api.h"
 #endif
 
+#include "arrow/acero/test_util_internal.h"
+
 #include "parquet/api/reader.h"
 #include "parquet/api/writer.h"
 
@@ -4884,6 +4887,7 @@ class TestArrowReadDictionary : public ::testing::TestWithParam<double> {
 
     std::shared_ptr<Table> actual;
     ASSERT_OK_NO_THROW(reader->ReadTable(&actual));
+
     ::arrow::AssertTablesEqual(expected, *actual, /*same_chunk_layout=*/false);
   }
 
@@ -5029,6 +5033,92 @@ INSTANTIATE_TEST_SUITE_P(
     ReadDictionary, TestArrowReadDictionary,
     ::testing::ValuesIn(TestArrowReadDictionary::null_probabilities()));
 
+class TestArrowReadRunEndEncoded : public ::testing::TestWithParam<double> {
+ public:
+  static constexpr int kNumRowGroups = 16;
+
+  struct {
+    int num_rows = 1024 * kNumRowGroups;
+    int num_row_groups = kNumRowGroups;
+    int num_uniques = 128;
+  } options;
+
+  void SetUp() override {
+    ASSERT_OK(::arrow::compute::Initialize());
+    properties_ = default_arrow_reader_properties();
+
+    GenerateData(GetParam());
+  }
+
+  void GenerateData(double null_probability) {
+    constexpr int64_t min_length = 2;
+    constexpr int64_t max_length = 100;
+    ::arrow::random::RandomArrayGenerator rag(0);
+    dense_values_ = rag.StringWithRepeats(options.num_rows, options.num_uniques,
+                                          min_length, max_length, null_probability);
+    expected_dense_ = MakeSimpleTable(dense_values_, /*nullable=*/true);
+  }
+
+  void TearDown() override {}
+
+  void WriteSimple() {
+    // Write `num_row_groups` row groups; each row group will have a different dictionary
+    ASSERT_NO_FATAL_FAILURE(
+        WriteTableToBuffer(expected_dense_, options.num_rows / options.num_row_groups,
+                           default_arrow_writer_properties(), &buffer_));
+  }
+
+  void CheckReadWholeFile(const Table& expected) {
+    ASSERT_OK_AND_ASSIGN(auto reader, GetReader());
+
+    std::shared_ptr<Table> actual;
+    ASSERT_OK_NO_THROW(reader->ReadTable(&actual));
+    ::arrow::AssertTablesEqual(expected, *actual, /*same_chunk_layout=*/false);
+  }
+
+  static std::vector<double> null_probabilities() { return {0.0, 0.5, 1}; }
+
+ protected:
+  std::shared_ptr<Array> dense_values_;
+  std::shared_ptr<Table> expected_dense_;
+  std::shared_ptr<Table> expected_dict_;
+  std::shared_ptr<Buffer> buffer_;
+  ArrowReaderProperties properties_;
+
+  ::arrow::Result<std::unique_ptr<FileReader>> GetReader() {
+    std::unique_ptr<FileReader> reader;
+
+    FileReaderBuilder builder;
+    RETURN_NOT_OK(builder.Open(std::make_shared<BufferReader>(buffer_)));
+    RETURN_NOT_OK(builder.properties(properties_)->Build(&reader));
+
+    return reader;
+  }
+};
+
+TEST_P(TestArrowReadRunEndEncoded, ReadWholeFile) {
+  properties_.set_read_ree(0, true);
+
+  WriteSimple();
+
+  auto num_row_groups = options.num_row_groups;
+  auto chunk_size = options.num_rows / num_row_groups;
+
+  std::vector<std::shared_ptr<Array>> chunks(num_row_groups);
+  for (int i = 0; i < num_row_groups; ++i) {
+    chunks[i] = dense_values_->Slice(chunk_size * i, chunk_size);
+  }
+  auto ex_table = MakeSimpleTable(std::make_shared<ChunkedArray>(chunks),
+                                  /*nullable=*/true);
+  ASSERT_OK_AND_ASSIGN(ex_table, ::arrow::acero::RunEndEncodeTableColumns(*ex_table, {0}));
+
+  CheckReadWholeFile(*ex_table);
+}
+
+INSTANTIATE_TEST_SUITE_P(
+    ReadRunEndEncoded, TestArrowReadRunEndEncoded,
+    ::testing::ValuesIn(TestArrowReadRunEndEncoded::null_probabilities()));
+
 TEST(TestArrowWriteDictionaries, ChangingDictionaries) {
   constexpr int num_unique = 50;
   constexpr int repeat = 10000;
diff --git a/cpp/src/parquet/arrow/reader.cc b/cpp/src/parquet/arrow/reader.cc
index d42fdc5034a..0c7cdcda7ec 100644
--- a/cpp/src/parquet/arrow/reader.cc
+++ b/cpp/src/parquet/arrow/reader.cc
@@ -462,7 +462,9 @@ class LeafReader : public ColumnReaderImpl {
     record_reader_ = RecordReader::Make(
         descr_, leaf_info, ctx_->pool,
         /*read_dictionary=*/field_->type()->id() == ::arrow::Type::DICTIONARY,
-        /*read_dense_for_nullable=*/false, /*arrow_type=*/type_for_reading);
+        /*read_dense_for_nullable=*/false,
+        /*read_run_end_encoded=*/field_->type()->id() == ::arrow::Type::RUN_END_ENCODED,
+        /*arrow_type=*/type_for_reading);
     NextRowGroup();
   }
 
diff --git a/cpp/src/parquet/arrow/reader_internal.cc b/cpp/src/parquet/arrow/reader_internal.cc
index b622e93e072..622e14ea7f8 100644
--- a/cpp/src/parquet/arrow/reader_internal.cc
+++ b/cpp/src/parquet/arrow/reader_internal.cc
@@ -95,6 +95,7 @@ using ::arrow::util::SafeLoadAs;
 using parquet::internal::BinaryRecordReader;
 using parquet::internal::DictionaryRecordReader;
 using parquet::internal::RecordReader;
+using parquet::internal::ReeRecordReader;
 using parquet::schema::GroupNode;
 using parquet::schema::Node;
 using parquet::schema::PrimitiveNode;
@@ -566,6 +567,24 @@ Status TransferDictionary(RecordReader* reader, MemoryPool* pool,
   return Status::OK();
 }
 
+Status TransferRunEndEncoded(RecordReader* reader, MemoryPool* pool,
+                             const std::shared_ptr<DataType>& logical_value_type,
+                             bool nullable, std::shared_ptr<ChunkedArray>* out) {
+  auto ree_reader = dynamic_cast<ReeRecordReader*>(reader);
+  DCHECK(ree_reader);
+  std::shared_ptr<::arrow::Array> result = ree_reader->GetResult();
+  *out = std::make_shared<ChunkedArray>(result);
+  if (!logical_value_type->Equals(*(*out)->type())) {
+    ARROW_ASSIGN_OR_RAISE(*out, ViewOrCastChunkedArray(*out, pool, logical_value_type));
+  }
+  if (!nullable) {
+    ::arrow::ArrayVector chunks = (*out)->chunks();
+    ReconstructChunksWithoutNulls(&chunks);
+    *out = std::make_shared<ChunkedArray>(std::move(chunks), logical_value_type);
+  }
+  return Status::OK();
+}
+
 Status TransferBinary(RecordReader* reader, MemoryPool* pool,
                       const std::shared_ptr<Field>& logical_type_field,
                       std::shared_ptr<ChunkedArray>* out) {
@@ -574,6 +593,12 @@ Status TransferBinary(RecordReader* reader, MemoryPool* pool,
         reader, pool, ::arrow::dictionary(::arrow::int32(), logical_type_field->type()),
         logical_type_field->nullable(), out);
   }
+  if (reader->read_ree()) {
+    return TransferRunEndEncoded(
+        reader, pool, ::arrow::run_end_encoded(::arrow::int32(), logical_type_field->type()),
+        logical_type_field->nullable(), out);
+  }
+
   ::arrow::compute::ExecContext ctx(pool);
   ::arrow::compute::CastOptions cast_options;
   cast_options.allow_invalid_utf8 = true;  // avoid spending time validating UTF8 data
@@ -864,6 +889,11 @@ Status TransferColumnData(RecordReader* reader,
                                        value_field->nullable(), &chunked_result));
       result = chunked_result;
     } break;
+    case ::arrow::Type::RUN_END_ENCODED: {
+      RETURN_NOT_OK(TransferRunEndEncoded(reader, pool, value_field->type(),
+                                          value_field->nullable(), &chunked_result));
+      result = chunked_result;
+    } break;
     case ::arrow::Type::NA: {
       result = std::make_shared<::arrow::NullArray>(reader->values_written());
       break;
diff --git a/cpp/src/parquet/arrow/schema.cc b/cpp/src/parquet/arrow/schema.cc
index 293ae94b94d..42ccc6652dd 100644
--- a/cpp/src/parquet/arrow/schema.cc
+++ b/cpp/src/parquet/arrow/schema.cc
@@ -537,6 +537,11 @@ bool IsDictionaryReadSupported(const ArrowType& type) {
   return type.id() == ::arrow::Type::BINARY || type.id() == ::arrow::Type::STRING;
 }
 
+bool IsRunEndEncodedReadSupported(const ArrowType& type) {
+  // Only supported currently for BYTE_ARRAY types
+  return type.id() == ::arrow::Type::BINARY || type.id() == ::arrow::Type::STRING;
+}
+
 // ----------------------------------------------------------------------
 // Schema logic
 
@@ -548,6 +553,9 @@ ::arrow::Result<std::shared_ptr<ArrowType>> GetTypeForNode(
   if (ctx->properties.read_dictionary(column_index) &&
       IsDictionaryReadSupported(*storage_type)) {
     return ::arrow::dictionary(::arrow::int32(), storage_type);
+  } else if (ctx->properties.read_ree(column_index) &&
+             IsRunEndEncodedReadSupported(*storage_type)) {
+    return ::arrow::run_end_encoded(::arrow::int32(), storage_type);
   }
   return storage_type;
 }
diff --git a/cpp/src/parquet/column_reader.cc b/cpp/src/parquet/column_reader.cc
index 8ecb774022f..a79fd1fc1ab 100644
--- a/cpp/src/parquet/column_reader.cc
+++ b/cpp/src/parquet/column_reader.cc
@@ -34,6 +34,7 @@
 #include "arrow/array/builder_binary.h"
 #include "arrow/array/builder_dict.h"
 #include "arrow/array/builder_primitive.h"
+#include "arrow/array/builder_run_end.h"
 #include "arrow/chunked_array.h"
 #include "arrow/type.h"
 #include "arrow/util/bit_stream_utils_internal.h"
@@ -2074,6 +2075,47 @@ class ByteArrayChunkedRecordReader final : public TypedRecordReader<ByteArrayTyp
   typename EncodingTraits<ByteArrayType>::Accumulator accumulator_;
 };
 
+class ByteArrayReeRecordReader final : public TypedRecordReader<ByteArrayType>,
+                                       virtual public ReeRecordReader {
+ public:
+  ByteArrayReeRecordReader(const ColumnDescriptor* descr, LevelInfo leaf_info,
+                           ::arrow::MemoryPool* pool, bool read_dense_for_nullable)
+      : TypedRecordReader<ByteArrayType>(descr, leaf_info, pool,
+                                         read_dense_for_nullable) {
+    read_ree_ = true;
+    ARROW_DCHECK_EQ(descr_->physical_type(), Type::BYTE_ARRAY);
+    PARQUET_THROW_NOT_OK(::arrow::MakeBuilder(
+        ::arrow::default_memory_pool(),
+        ::arrow::run_end_encoded(::arrow::int32(), ::arrow::binary()), &builder_));
+  }
+
+  void ReadValuesDense(int64_t values_to_read) override {
+    int64_t num_decoded = this->current_decoder_->DecodeArrowNonNull(
+        static_cast<int>(values_to_read),
+        checked_cast<::arrow::RunEndEncodedBuilder*>(builder_.get()));
+    CheckNumberDecoded(num_decoded, values_to_read);
+    ResetValues();
+  }
+
+  void ReadValuesSpaced(int64_t values_to_read, int64_t null_count) override {
+    int64_t num_decoded = this->current_decoder_->DecodeArrow(
+        static_cast<int>(values_to_read), static_cast<int>(null_count),
+        valid_bits_->mutable_data(), values_written_,
+        checked_cast<::arrow::RunEndEncodedBuilder*>(builder_.get()));
+    CheckNumberDecoded(num_decoded, values_to_read - null_count);
+    ResetValues();
+  }
+
+  std::shared_ptr<::arrow::Array> GetResult() override {
+    std::shared_ptr<::arrow::Array> result;
+    PARQUET_THROW_NOT_OK(builder_->Finish(&result));
+    return result;
+  }
+
+ private:
+  std::unique_ptr<::arrow::ArrayBuilder> builder_;
+};
+
 /// ByteArrayDictionaryRecordReader reads into ::arrow::dictionary(index: int32,
 /// values: binary).
 ///
@@ -2171,11 +2213,14 @@ void TypedRecordReader<FLBAType>::DebugPrintState() {}
 
 std::shared_ptr<RecordReader> MakeByteArrayRecordReader(
     const ColumnDescriptor* descr, LevelInfo leaf_info, ::arrow::MemoryPool* pool,
-    bool read_dictionary, bool read_dense_for_nullable,
+    bool read_dictionary, bool read_dense_for_nullable, bool read_run_end_encoded,
     const std::shared_ptr<::arrow::DataType>& arrow_type) {
   if (read_dictionary) {
     return std::make_shared<ByteArrayDictionaryRecordReader>(descr, leaf_info, pool,
                                                              read_dense_for_nullable);
+  } else if (read_run_end_encoded) {
+    return std::make_shared<ByteArrayReeRecordReader>(
+        descr, leaf_info, pool, read_dense_for_nullable);
   } else {
     return std::make_shared<ByteArrayChunkedRecordReader>(
         descr, leaf_info, pool, read_dense_for_nullable, arrow_type);
@@ -2186,7 +2231,7 @@ std::shared_ptr<RecordReader> MakeByteArrayRecordReader(
 
 std::shared_ptr<RecordReader> RecordReader::Make(
     const ColumnDescriptor* descr, LevelInfo leaf_info, MemoryPool* pool,
-    bool read_dictionary, bool read_dense_for_nullable,
+    bool read_dictionary, bool read_dense_for_nullable, bool read_run_end_encoded,
     const std::shared_ptr<::arrow::DataType>& arrow_type) {
   switch (descr->physical_type()) {
     case Type::BOOLEAN:
@@ -2207,10 +2252,9 @@ std::shared_ptr<RecordReader> RecordReader::Make(
     case Type::DOUBLE:
       return std::make_shared<TypedRecordReader<DoubleType>>(descr, leaf_info, pool,
                                                              read_dense_for_nullable);
-    case Type::BYTE_ARRAY: {
+    case Type::BYTE_ARRAY:
       return MakeByteArrayRecordReader(descr, leaf_info, pool, read_dictionary,
-                                       read_dense_for_nullable, arrow_type);
-    }
+                                       read_dense_for_nullable, read_run_end_encoded, arrow_type);
     case Type::FIXED_LEN_BYTE_ARRAY:
       return std::make_shared<FLBARecordReader>(descr, leaf_info, pool,
                                                 read_dense_for_nullable);
diff --git a/cpp/src/parquet/column_reader.h b/cpp/src/parquet/column_reader.h
index ac4469b1904..41453d41d04 100644
--- a/cpp/src/parquet/column_reader.h
+++ b/cpp/src/parquet/column_reader.h
@@ -273,6 +273,7 @@ class PARQUET_EXPORT RecordReader {
       const ColumnDescriptor* descr, LevelInfo leaf_info,
       ::arrow::MemoryPool* pool = ::arrow::default_memory_pool(),
       bool read_dictionary = false, bool read_dense_for_nullable = false,
+      bool read_run_end_encoded = false,
       const std::shared_ptr<::arrow::DataType>& arrow_type = NULLPTR);
 
   virtual ~RecordReader() = default;
@@ -373,6 +374,9 @@ class PARQUET_EXPORT RecordReader {
   /// \brief True if reading directly as Arrow dictionary-encoded
   bool read_dictionary() const { return read_dictionary_; }
 
+  /// \brief True if reading directly as Arrow run-end-encoded
+  bool read_ree() const { return read_ree_; }
+
   /// \brief True if reading dense for nullable columns.
   bool read_dense_for_nullable() const { return read_dense_for_nullable_; }
 
@@ -424,6 +428,7 @@ class PARQUET_EXPORT RecordReader {
   int64_t levels_capacity_;
 
   bool read_dictionary_ = false;
+  bool read_ree_ = false;
   // If true, we will not leave any space for the null values in the values_
   // vector or fill nulls values in BinaryRecordReader/DictionaryRecordReader.
   //
@@ -444,6 +449,13 @@ class DictionaryRecordReader : virtual public RecordReader {
   virtual std::shared_ptr<::arrow::ChunkedArray> GetResult() = 0;
 };
 
+/// \brief Read records directly to run-end-encoded Arrow form (int32
+/// run ends). Only valid for BYTE_ARRAY columns
+class ReeRecordReader : virtual public RecordReader {
+ public:
+  virtual std::shared_ptr<::arrow::Array> GetResult() = 0;
+};
+
 }  // namespace internal
 
 using BoolReader = TypedColumnReader<BooleanType>;
diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc
index d0a857dd22a..ae3cbf71f07 100644
--- a/cpp/src/parquet/decoder.cc
+++ b/cpp/src/parquet/decoder.cc
@@ -33,6 +33,7 @@
 #include "arrow/array/builder_binary.h"
 #include "arrow/array/builder_dict.h"
 #include "arrow/array/builder_primitive.h"
+#include "arrow/builder.h"
 #include "arrow/type_traits.h"
 #include "arrow/util/bit_block_counter.h"
 #include "arrow/util/bit_run_reader.h"
@@ -272,6 +273,45 @@ void CheckPageLargeEnough(int64_t remaining_bytes, int32_t value_width,
   }
 }
 
+
+class ReeBuilderHelper {
+ public:
+  ReeBuilderHelper(typename EncodingTraits<ByteArrayType>::ReeAccumulator* builder)
+      : builder_(builder),
+        arrow_value_dtype_(
+            std::static_pointer_cast<::arrow::RunEndEncodedType>(builder_->type())
+                ->value_type()) {};
+
+  Status flushCurrStateToBuilder() {
+    if (curr_val_.has_value()) {
+      auto str_value = ByteArrayToString(*curr_val_);
+      ARROW_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Scalar> scalar,
+                            ::arrow::MakeScalar(arrow_value_dtype_, str_value));
+      RETURN_NOT_OK(builder_->AppendScalar(*scalar, curr_repeats_));
+    } else if (curr_repeats_ > 0) {
+      RETURN_NOT_OK(builder_->AppendNulls(curr_repeats_));
+    }
+    return Status::OK();
+  }
+
+  Status update(std::optional<parquet::ByteArray> val, int num_repeats) {
+    if (val == curr_val_) {
+      curr_repeats_ += num_repeats;
+    } else {
+      RETURN_NOT_OK(flushCurrStateToBuilder());
+      curr_val_ = val;
+      curr_repeats_ = num_repeats;
+    }
+    return Status::OK();
+  }
+
+ private:
+  std::optional<parquet::ByteArray> curr_val_;
+  int curr_repeats_ = 0;
+  typename EncodingTraits<ByteArrayType>::ReeAccumulator* builder_;
+  std::shared_ptr<::arrow::DataType> arrow_value_dtype_;
+};
+
 // Internal decoder class hierarchy
 
 class DecoderImpl : virtual public Decoder {
@@ -359,6 +399,10 @@ class PlainDecoder : public TypedDecoderImpl<DType> {
   int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
                   int64_t valid_bits_offset,
                   typename EncodingTraits<DType>::DictAccumulator* builder) override;
+
+  int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
+                  int64_t valid_bits_offset,
+                  typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) override;
 };
 
 template <>
@@ -448,6 +492,13 @@ int PlainDecoder<DType>::DecodeArrow(
   return values_decoded;
 }
 
+template <typename DType>
+inline int PlainDecoder<DType>::DecodeArrow(
+    int num_values, int null_count, const uint8_t* valid_bits, int64_t valid_bits_offset,
+    typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) {
+  ParquetException::NYI("run-end-encoded PlainDecoder");
+}
+
 // Decode routine templated on C++ type rather than type enum
 template <typename T>
 inline int DecodePlain(const uint8_t* data, int64_t data_size, int num_values,
@@ -544,6 +595,10 @@ class PlainBooleanDecoder : public TypedDecoderImpl<BooleanType>, public Boolean
                   int64_t valid_bits_offset,
                   typename EncodingTraits<BooleanType>::DictAccumulator* out) override;
 
+  int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
+                  int64_t valid_bits_offset,
+                  typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) override;
+
  private:
   std::unique_ptr<::arrow::bit_util::BitReader> bit_reader_;
   int total_num_values_{0};
@@ -621,6 +676,12 @@ inline int PlainBooleanDecoder::DecodeArrow(
   ParquetException::NYI("dictionaries of BooleanType");
 }
 
+inline int PlainBooleanDecoder::DecodeArrow(
+    int num_values, int null_count, const uint8_t* valid_bits, int64_t valid_bits_offset,
+    typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) {
+  ParquetException::NYI("run-end-encoded BooleanType");
+}
+
 int PlainBooleanDecoder::Decode(uint8_t* buffer, int max_values) {
   max_values = std::min(max_values, num_values_);
   if (ARROW_PREDICT_FALSE(!bit_reader_->Advance(max_values))) {
@@ -749,6 +810,15 @@ class PlainByteArrayDecoder : public PlainDecoder<ByteArrayType> {
     return result;
   }
 
+  int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
+                  int64_t valid_bits_offset,
+                  typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) override {
+    int result = 0;
+    PARQUET_THROW_NOT_OK(DecodeArrowDense(num_values, null_count, valid_bits,
+                                          valid_bits_offset, out, &result));
+    return result;
+  }
+
  private:
   Status DecodeArrowDense(int num_values, int null_count, const uint8_t* valid_bits,
                           int64_t valid_bits_offset,
@@ -794,6 +864,46 @@ class PlainByteArrayDecoder : public PlainDecoder<ByteArrayType> {
                                                     visit_binary_helper);
   }
 
+  Status DecodeArrowDense(int num_values, int null_count, const uint8_t* valid_bits,
+                          int64_t valid_bits_offset,
+                          typename EncodingTraits<ByteArrayType>::ReeAccumulator* out,
+                          int* out_values_decoded) {
+    ReeBuilderHelper helper(out);
+    int values_decoded = 0;
+    int i = 0;
+    RETURN_NOT_OK(VisitNullBitmapInline(
+        valid_bits, valid_bits_offset, num_values, null_count,
+        [&]() {
+          if (ARROW_PREDICT_FALSE(len_ < 4)) {
+            ParquetException::EofException();
+          }
+          auto value_len = SafeLoadAs<int32_t>(data_);
+          if (ARROW_PREDICT_FALSE(value_len < 0 || value_len > INT32_MAX - 4)) {
+            return Status::Invalid("Invalid or corrupted value_len '", value_len, "'");
+          }
+          auto increment = value_len + 4;
+          if (ARROW_PREDICT_FALSE(len_ < increment)) {
+            ParquetException::EofException();
+          }
+          parquet::ByteArray byte_array(value_len, data_ + 4);
+          RETURN_NOT_OK(helper.update(byte_array, 1));
+          data_ += increment;
+          len_ -= increment;
+          ++values_decoded;
+          ++i;
+          return Status::OK();
+        },
+        [&]() {
+          RETURN_NOT_OK(helper.update(std::nullopt, 1));
+          ++i;
+          return Status::OK();
+        }));
+    RETURN_NOT_OK(helper.flushCurrStateToBuilder());
+    num_values_ -= values_decoded;
+    *out_values_decoded = values_decoded;
+    return Status::OK();
+  }
+
   template <typename BuilderType>
   Status DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
                      int64_t valid_bits_offset, BuilderType* builder,
@@ -909,6 +1019,13 @@ class DictDecoderImpl : public TypedDecoderImpl<Type>, public DictDecoder<Type>
                   int64_t valid_bits_offset,
                   typename EncodingTraits<Type>::DictAccumulator* out) override;
 
+  int DecodeArrow(
+      int num_values, int null_count, const uint8_t* valid_bits,
+      int64_t valid_bits_offset,
+      typename EncodingTraits<ByteArrayType>::ReeAccumulator* builder) override {
+    ParquetException::NYI();
+  }
+
   void InsertDictionary(::arrow::ArrayBuilder* builder) override;
 
   int DecodeIndicesSpaced(int num_values, int null_count, const uint8_t* valid_bits,
@@ -1271,7 +1388,81 @@ class DictByteArrayDecoderImpl : public DictDecoderImpl<ByteArrayType> {
     return result;
   }
 
+  int DecodeArrow(
+      int num_values, int null_count, const uint8_t* valid_bits,
+      int64_t valid_bits_offset,
+      typename EncodingTraits<ByteArrayType>::ReeAccumulator* builder) override {
+    int result = 0;
+    if (null_count == 0) {
+      PARQUET_THROW_NOT_OK(DecodeArrowDenseNonNull(num_values, builder, &result));
+    } else {
+      PARQUET_THROW_NOT_OK(DecodeArrowDense(num_values, null_count, valid_bits,
+                                            valid_bits_offset, builder, &result));
+    }
+    return result;
+  }
+
  private:
+  Status DecodeArrowDense(int num_values, int null_count, const uint8_t* valid_bits,
+                          int64_t valid_bits_offset,
+                          typename EncodingTraits<ByteArrayType>::ReeAccumulator* builder,
+                          int* out_num_values) {
+    int total_values_decoded = 0;
+    int non_null_values_decoded = 0;
+    ReeBuilderHelper helper(builder);
+    const auto* dict_values = dictionary_->data_as<ByteArray>();
+    int64_t valid_bits_index = valid_bits_offset;
+    while (total_values_decoded < num_values) {
+      int32_t idx;
+      bool is_null = false;
+      int num_repeats;
+      bool ok = idx_decoder_.GetNextValueAndNumRepeatsSpaced(&idx, &is_null, &num_repeats,
+                                                     num_values - total_values_decoded,
+                                                     valid_bits, valid_bits_index);
+      if (ARROW_PREDICT_FALSE(!ok)) {
+        break;
+      }
+      DCHECK_GT(num_repeats, 0);
+      if (is_null) {
+        RETURN_NOT_OK(helper.update(std::nullopt, num_repeats));
+      } else {
+        RETURN_NOT_OK(IndexInBounds(idx));
+        const auto& val = dict_values[idx];
+        RETURN_NOT_OK(helper.update(val, num_repeats));
+        non_null_values_decoded += num_repeats;
+      }
+      valid_bits_index += num_repeats;
+      total_values_decoded += num_repeats;
+    }
+    RETURN_NOT_OK(helper.flushCurrStateToBuilder());
+    *out_num_values = non_null_values_decoded;
+    return Status::OK();
+  }
+
+  Status DecodeArrowDenseNonNull(
+      int num_values, typename EncodingTraits<ByteArrayType>::ReeAccumulator* builder,
+      int* out_num_values) {
+    int values_decoded = 0;
+    ReeBuilderHelper helper(builder);
+    const auto* dict_values = dictionary_->data_as<ByteArray>();
+    while (values_decoded < num_values) {
+      int32_t idx;
+      int num_repeats;
+      bool ok = idx_decoder_.GetNextValueAndNumRepeats(&idx, &num_repeats, num_values - values_decoded);
+      if (ARROW_PREDICT_FALSE(!ok)) {
+        break;
+      }
+      DCHECK_GT(num_repeats, 0);
+      RETURN_NOT_OK(IndexInBounds(idx));
+      const auto& val = dict_values[idx];
+      RETURN_NOT_OK(helper.update(val, num_repeats));
+      values_decoded += num_repeats;
+    }
+    RETURN_NOT_OK(helper.flushCurrStateToBuilder());
+    *out_num_values = values_decoded;
+    return Status::OK();
+  }
+
   Status DecodeArrowDense(int num_values, int null_count, const uint8_t* valid_bits,
                           int64_t valid_bits_offset,
                           typename EncodingTraits<ByteArrayType>::Accumulator* out,
@@ -1486,6 +1677,12 @@ class DeltaBitPackDecoder : public TypedDecoderImpl<DType> {
     return decoded_count;
   }
 
+  int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
+                  int64_t valid_bits_offset,
+                  typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) override {
+    ParquetException::NYI("Run end encoded DeltaBitPackDecoder");
+  }
+
  private:
   static constexpr int kMaxDeltaBitWidth = static_cast<int>(sizeof(T) * 8);
 
@@ -1728,6 +1925,12 @@ class DeltaLengthByteArrayDecoder : public TypedDecoderImpl<ByteArrayType> {
         "DecodeArrow of DictAccumulator for DeltaLengthByteArrayDecoder");
   }
 
+  int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
+                  int64_t valid_bits_offset,
+                  typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) override {
+    ParquetException::NYI("DecodeArrow of ReeAccumulator for DeltaLengthByteArrayDecoder");
+  }
+
  private:
   // Decode all the encoded lengths. The decoder_ will be at the start of the encoded data
   // after that.
@@ -1901,7 +2104,13 @@ class RleBooleanDecoder : public TypedDecoderImpl<BooleanType>, public BooleanDe
       int num_values, int null_count, const uint8_t* valid_bits,
       int64_t valid_bits_offset,
       typename EncodingTraits<BooleanType>::DictAccumulator* builder) override {
-    ParquetException::NYI("DecodeArrow for RleBooleanDecoder");
+    ParquetException::NYI("DecodeArrow of DictAccumulator for RleBooleanDecoder");
+  }
+
+  int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
+                  int64_t valid_bits_offset,
+                  typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) override {
+    ParquetException::NYI("DecodeArrow of ReeAccumulator for RleBooleanDecoder");
   }
 
  private:
@@ -1973,6 +2182,12 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl<DType> {
     ParquetException::NYI("DecodeArrow of DictAccumulator for DeltaByteArrayDecoder");
   }
 
+  int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
+                  int64_t valid_bits_offset,
+                  typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) override {
+    ParquetException::NYI("DecodeArrow of ReeAccumulator for DeltaByteArrayDecoder");
+  }
+
  protected:
   template <bool is_first_run>
   static void BuildBufferInternal(const int32_t* prefix_len_ptr, int i, ByteArray* buffer,
@@ -2211,6 +2426,12 @@ class ByteStreamSplitDecoderBase : public TypedDecoderImpl<DType> {
     ParquetException::NYI("DecodeArrow to DictAccumulator for BYTE_STREAM_SPLIT");
   }
 
+  int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
+                  int64_t valid_bits_offset,
+                  typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) override {
+    ParquetException::NYI("DecodeArrow to ReeAccumulator for BYTE_STREAM_SPLIT");
+  }
+
   int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits,
                   int64_t valid_bits_offset,
                   typename EncodingTraits<DType>::Accumulator* builder) override {
@@ -2241,6 +2462,7 @@ class ByteStreamSplitDecoderBase : public TypedDecoderImpl<DType> {
     return values_to_decode;
   }
 
+
  protected:
   int DecodeRaw(uint8_t* out_buffer, int max_values) {
     const int values_to_decode = std::min(this->num_values_, max_values);
diff --git a/cpp/src/parquet/encoding.h b/cpp/src/parquet/encoding.h
index d80bf0edcae..e900c5f11cf 100644
--- a/cpp/src/parquet/encoding.h
+++ b/cpp/src/parquet/encoding.h
@@ -70,6 +70,7 @@ struct EncodingTraits<BooleanType> {
   using ArrowType = ::arrow::BooleanType;
   using Accumulator = ::arrow::BooleanBuilder;
   struct DictAccumulator {};
+  struct ReeAccumulator {};
 };
 
 template <>
@@ -80,6 +81,7 @@ struct EncodingTraits<Int32Type> {
   using ArrowType = ::arrow::Int32Type;
   using Accumulator = ::arrow::NumericBuilder<::arrow::Int32Type>;
   using DictAccumulator = ::arrow::Dictionary32Builder<::arrow::Int32Type>;
+  struct ReeAccumulator {};
 };
 
 template <>
@@ -90,6 +92,7 @@ struct EncodingTraits<Int64Type> {
   using ArrowType = ::arrow::Int64Type;
   using Accumulator = ::arrow::NumericBuilder<::arrow::Int64Type>;
   using DictAccumulator = ::arrow::Dictionary32Builder<::arrow::Int64Type>;
+  struct ReeAccumulator {};
 };
 
 template <>
@@ -99,6 +102,7 @@ struct EncodingTraits<Int96Type> {
 
   struct Accumulator {};
   struct DictAccumulator {};
+  struct ReeAccumulator {};
 };
 
 template <>
@@ -109,6 +113,7 @@ struct EncodingTraits<FloatType> {
   using ArrowType = ::arrow::FloatType;
   using Accumulator = ::arrow::NumericBuilder<::arrow::FloatType>;
   using DictAccumulator = ::arrow::Dictionary32Builder<::arrow::FloatType>;
+  struct ReeAccumulator {};
 };
 
 template <>
@@ -119,6 +124,7 @@ struct EncodingTraits<DoubleType> {
   using ArrowType = ::arrow::DoubleType;
   using Accumulator = ::arrow::NumericBuilder<::arrow::DoubleType>;
   using DictAccumulator = ::arrow::Dictionary32Builder<::arrow::DoubleType>;
+  struct ReeAccumulator {};
 };
 
 template <>
@@ -139,6 +145,7 @@ struct EncodingTraits<ByteArrayType> {
     std::vector<std::shared_ptr<::arrow::Array>> chunks;
   };
   using DictAccumulator = ::arrow::Dictionary32Builder<::arrow::BinaryType>;
+  using ReeAccumulator = ::arrow::RunEndEncodedBuilder;
 };
 
 template <>
@@ -149,6 +156,7 @@ struct EncodingTraits<FLBAType> {
   using ArrowType = ::arrow::FixedSizeBinaryType;
   using Accumulator = ::arrow::FixedSizeBinaryBuilder;
   using DictAccumulator = ::arrow::Dictionary32Builder<::arrow::FixedSizeBinaryType>;
+  struct ReeAccumulator {};
 };
 
 class ColumnDescriptor;
@@ -324,6 +332,11 @@ class TypedDecoder : virtual public Decoder {
                           int64_t valid_bits_offset,
                           typename EncodingTraits<DType>::DictAccumulator* builder) = 0;
 
+  virtual int DecodeArrow(
+      int num_values, int null_count, const uint8_t* valid_bits,
+      int64_t valid_bits_offset,
+      typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) = 0;
+
   /// \brief Decode into a DictionaryBuilder ignoring nulls
   ///
   /// \return number of values decoded
@@ -331,6 +344,14 @@ class TypedDecoder : virtual public Decoder {
                          typename EncodingTraits<DType>::DictAccumulator* builder) {
     return DecodeArrow(num_values, 0, /*valid_bits=*/NULLPTR, 0, builder);
   }
+
+  /// \brief Decode into a RunEndEncodedBuilder ignoring nulls
+  ///
+  /// \return number of values decoded
+  int DecodeArrowNonNull(int num_values,
+                         typename EncodingTraits<ByteArrayType>::ReeAccumulator* out) {
+    return DecodeArrow(num_values, 0, /*valid_bits=*/NULLPTR, 0, out);
+  }
 };
 
 template <typename DType>
diff --git a/cpp/src/parquet/file_reader.cc b/cpp/src/parquet/file_reader.cc
index b246feaf732..9b245d11ea3 100644
--- a/cpp/src/parquet/file_reader.cc
+++ b/cpp/src/parquet/file_reader.cc
@@ -111,7 +111,7 @@ std::shared_ptr<ColumnReader> RowGroupReader::Column(int i) {
 }
 
 std::shared_ptr<internal::RecordReader> RowGroupReader::RecordReader(
-    int i, bool read_dictionary) {
+    int i, bool read_dictionary, bool read_ree) {
   if (i >= metadata()->num_columns()) {
     std::stringstream ss;
     ss << "Trying to read column index " << i << " but row group metadata has only "
@@ -126,7 +126,7 @@ std::shared_ptr<internal::RecordReader> RowGroupReader::RecordReader(
 
   auto reader = internal::RecordReader::Make(
       descr, level_info, contents_->properties()->memory_pool(), read_dictionary,
-      contents_->properties()->read_dense_for_nullable());
+      contents_->properties()->read_dense_for_nullable(), read_ree);
   reader->SetPageReader(std::move(page_reader));
   return reader;
 }
@@ -149,7 +149,8 @@ std::shared_ptr<internal::RecordReader> RowGroupReader::RecordReaderWithExposeEn
   return RecordReader(
       i,
       /*read_dictionary=*/encoding_to_expose == ExposedEncoding::DICTIONARY &&
-          IsColumnChunkFullyDictionaryEncoded(*metadata()->ColumnChunk(i)));
+          IsColumnChunkFullyDictionaryEncoded(*metadata()->ColumnChunk(i)),
+      /*read_ree=*/encoding_to_expose == ExposedEncoding::REE);
 }
 
 std::unique_ptr<PageReader> RowGroupReader::GetColumnPageReader(int i) {
diff --git a/cpp/src/parquet/file_reader.h b/cpp/src/parquet/file_reader.h
index c42163276cd..2da052a196e 100644
--- a/cpp/src/parquet/file_reader.h
+++ b/cpp/src/parquet/file_reader.h
@@ -65,7 +65,8 @@ class PARQUET_EXPORT RowGroupReader {
   // EXPERIMENTAL: Construct a RecordReader for the indicated column of the row group.
   // Ownership is shared with the RowGroupReader.
   std::shared_ptr<internal::RecordReader> RecordReader(int i,
-                                                       bool read_dictionary = false);
+                                                       bool read_dictionary = false,
+                                                       bool read_ree = false);
 
   // Construct a ColumnReader, trying to enable exposed encoding.
   //
@@ -83,7 +84,7 @@ class PARQUET_EXPORT RowGroupReader {
 
   // Construct a RecordReader, trying to enable exposed encoding.
   //
-  // For dictionary encoding, currently we only support column chunks that are
+  // For dictionary and run-end encoding, currently we only support column chunks that are
   // fully dictionary encoded byte arrays. The caller should verify if the reader can read
   // and expose the dictionary by checking the reader's read_dictionary(). If a column
   // chunk uses dictionary encoding but then falls back to plain encoding, the returned
diff --git a/cpp/src/parquet/properties.h b/cpp/src/parquet/properties.h
index 5a1799c39d7..618bd2f3633 100644
--- a/cpp/src/parquet/properties.h
+++ b/cpp/src/parquet/properties.h
@@ -1038,6 +1038,25 @@ class PARQUET_EXPORT ArrowReaderProperties {
     }
   }
 
+  /// \brief Set whether to read a particular column as run-end-encoded.
+  ///
+  /// This is only supported for columns with a Parquet physical type of
+  /// BYTE_ARRAY, such as string or binary types.
+  void set_read_ree(int column_index, bool read_ree) {
+    if (read_ree) {
+      read_ree_indices_.insert(column_index);
+    } else {
+      read_ree_indices_.erase(column_index);
+    }
+  }
+  bool read_ree(int column_index) const {
+    if (read_ree_indices_.find(column_index) != read_ree_indices_.end()) {
+      return true;
+    } else {
+      return false;
+    }
+  }
+
   /// \brief Set the Arrow binary type to read BYTE_ARRAY columns as.
   ///
   /// Allowed values are Type::BINARY, Type::LARGE_BINARY and Type::BINARY_VIEW.
@@ -1148,6 +1167,7 @@ class PARQUET_EXPORT ArrowReaderProperties {
  private:
   bool use_threads_;
   std::unordered_set<int> read_dict_indices_;
+  std::unordered_set<int> read_ree_indices_;
   int64_t batch_size_;
   bool pre_buffer_;
   ::arrow::io::IOContext io_context_;
diff --git a/cpp/src/parquet/reader_test.cc b/cpp/src/parquet/reader_test.cc
index 7ae9021e35e..34ec8517258 100644
--- a/cpp/src/parquet/reader_test.cc
+++ b/cpp/src/parquet/reader_test.cc
@@ -37,6 +37,8 @@
 #include "arrow/array/array_binary.h"
 #include "arrow/array/builder_binary.h"
 #include "arrow/buffer.h"
+#include "arrow/compute/api_vector.h"
+#include "arrow/datum.h"
 #include "arrow/io/file.h"
 #include "arrow/testing/future_util.h"
 #include "arrow/testing/gtest_util.h"
@@ -712,6 +714,74 @@ TEST(TestFileReader, RecordReaderWithExposingDictionary) {
   }
 }
 
+TEST(TestFileReader, RecordReaderWithExposingRee) {
+  const int num_rows = 1000;
+
+  // Make schema
+  schema::NodeVector fields;
+  fields.push_back(PrimitiveNode::Make("field", Repetition::REQUIRED, Type::BYTE_ARRAY,
+                                       ConvertedType::NONE));
+  auto schema = std::static_pointer_cast<GroupNode>(
+      GroupNode::Make("schema", Repetition::REQUIRED, fields));
+
+  // Write small batches and small data pages
+  std::shared_ptr<WriterProperties> writer_props = WriterProperties::Builder()
+                                                       .write_batch_size(64)
+                                                       ->data_pagesize(128)
+                                                       ->enable_dictionary()
+                                                       ->build();
+
+  ASSERT_OK_AND_ASSIGN(auto out_file, ::arrow::io::BufferOutputStream::Create());
+  std::shared_ptr<ParquetFileWriter> file_writer =
+      ParquetFileWriter::Open(out_file, schema, writer_props);
+
+  RowGroupWriter* rg_writer = file_writer->AppendRowGroup();
+
+  // write one column
+  ::arrow::random::RandomArrayGenerator rag(0);
+  ByteArrayWriter* writer = static_cast<ByteArrayWriter*>(rg_writer->NextColumn());
+  std::vector<std::string> raw_unique_data = {"a", "bc", "defg"};
+  std::vector<ByteArray> col_typed;
+  for (int i = 0; i < num_rows; i++) {
+    std::string_view chosed_data = raw_unique_data[i % raw_unique_data.size()];
+    col_typed.emplace_back(chosed_data);
+  }
+  writer->WriteBatch(num_rows, nullptr, nullptr, col_typed.data());
+  rg_writer->Close();
+  file_writer->Close();
+
+  // Open the reader
+  ASSERT_OK_AND_ASSIGN(auto file_buf, out_file->Finish());
+  auto in_file = std::make_shared<::arrow::io::BufferReader>(file_buf);
+
+  ReaderProperties reader_props;
+  reader_props.enable_buffered_stream();
+  reader_props.set_buffer_size(64);
+  std::unique_ptr<ParquetFileReader> file_reader =
+      ParquetFileReader::Open(in_file, reader_props);
+
+  auto row_group = file_reader->RowGroup(0);
+  auto record_reader = std::dynamic_pointer_cast<internal::ReeRecordReader>(
+      row_group->RecordReaderWithExposeEncoding(0, ExposedEncoding::REE));
+  ASSERT_NE(record_reader, nullptr);
+  ASSERT_TRUE(record_reader->read_ree());
+
+  ASSERT_EQ(record_reader->ReadRecords(num_rows), num_rows);
+  std::shared_ptr<::arrow::Array> ree_result_array = record_reader->GetResult();
+  ASSERT_OK_AND_ASSIGN(::arrow::Datum decoded_datum,
+      ::arrow::compute::RunEndDecode(::arrow::Datum(ree_result_array)));
+  auto decoded_array = decoded_datum.make_array();
+  auto decoded_binary_array = std::static_pointer_cast<::arrow::BinaryArray>(decoded_array);
+
+  // Verify values using RunEndDecode
+  int64_t indices_read = decoded_binary_array->length();
+  ASSERT_EQ(indices_read, num_rows);
+  for (int i = 0; i < indices_read; ++i) {
+    std::string_view ith_value_view = decoded_binary_array->GetView(i);
+    ASSERT_EQ(ith_value_view, col_typed[i]);
+  }
+}
+
 class TestLocalFile : public ::testing::Test {
  public:
   void SetUp() {
diff --git a/cpp/src/parquet/types.h b/cpp/src/parquet/types.h
index 7e8a18fc94d..5ab973a301f 100644
--- a/cpp/src/parquet/types.h
+++ b/cpp/src/parquet/types.h
@@ -550,7 +550,8 @@ struct Encoding {
 // decoding, in which case the data read from the file is DICTIONARY encoded.
 enum class ExposedEncoding {
   NO_ENCODING = 0,  // data is not encoded, i.e. already decoded during reading
-  DICTIONARY = 1
+  DICTIONARY = 1,
+  REE = 2
 };
 
 /// \brief Return true if Parquet supports indicated compression type
diff --git a/python/pyarrow/_dataset_parquet.pyx b/python/pyarrow/_dataset_parquet.pyx
index 9405b5d8c54..61bb6124533 100644
--- a/python/pyarrow/_dataset_parquet.pyx
+++ b/python/pyarrow/_dataset_parquet.pyx
@@ -142,6 +142,9 @@ cdef class ParquetFileFormat(FileFormat):
         if read_options.dictionary_columns is not None:
             for column in read_options.dictionary_columns:
                 options.dict_columns.insert(tobytes(column))
+        if read_options.ree_columns is not None:
+            for column in read_options.ree_columns:
+                options.ree_columns.insert(tobytes(column))
         options.coerce_int96_timestamp_unit = \
             read_options._coerce_int96_timestamp_unit
         options.binary_type = read_options._binary_type
@@ -181,6 +184,7 @@ cdef class ParquetFileFormat(FileFormat):
         parquet_read_options = ParquetReadOptions(
             dictionary_columns={frombytes(col)
                                 for col in options.dict_columns},
+            ree_columns={frombytes(col) for col in options.ree_columns},
         )
         # Read options getter/setter works with strings so setting
         # the private property which uses the C Type
@@ -527,15 +531,18 @@ cdef class ParquetReadOptions(_Weakrefable):
 
     cdef public:
         set dictionary_columns
+        set ree_columns
         TimeUnit _coerce_int96_timestamp_unit
         Type _binary_type
         Type _list_type
 
     # Also see _PARQUET_READ_OPTIONS
     def __init__(self, dictionary_columns=None,
+                 ree_columns=None,
                  coerce_int96_timestamp_unit=None,
                  binary_type=None, list_type=None):
         self.dictionary_columns = set(dictionary_columns or set())
+        self.ree_columns = set(ree_columns or set())
         self.coerce_int96_timestamp_unit = coerce_int96_timestamp_unit
         self.binary_type = binary_type
         self.list_type = list_type
@@ -585,6 +592,7 @@ cdef class ParquetReadOptions(_Weakrefable):
         bool
         """
         return (self.dictionary_columns == other.dictionary_columns and
+                self.ree_columns == other.ree_columns and
                 self._coerce_int96_timestamp_unit ==
                 other._coerce_int96_timestamp_unit and
                 self._binary_type == other._binary_type and
@@ -600,6 +608,7 @@ cdef class ParquetReadOptions(_Weakrefable):
         return (
             f"<ParquetReadOptions"
             f" dictionary_columns={self.dictionary_columns}"
+            f" ree_columns={self.ree_columns}"
             f" coerce_int96_timestamp_unit={self.coerce_int96_timestamp_unit}"
             f" binary_type={self.binary_type}"
             f" list_type={self.list_type}"
@@ -721,7 +730,7 @@ cdef class ParquetFileWriteOptions(FileWriteOptions):
 
 
 cdef set _PARQUET_READ_OPTIONS = {
-    'dictionary_columns', 'coerce_int96_timestamp_unit', 'binary_type', 'list_type',
+    'dictionary_columns', 'ree_columns', 'coerce_int96_timestamp_unit', 'binary_type', 'list_type',
 }
 
 
diff --git a/python/pyarrow/_parquet.pyx b/python/pyarrow/_parquet.pyx
index 14cd3e363a4..4d5d1548ed0 100644
--- a/python/pyarrow/_parquet.pyx
+++ b/python/pyarrow/_parquet.pyx
@@ -1571,7 +1571,8 @@ cdef class ParquetReader(_Weakrefable):
              thrift_string_size_limit=None,
              thrift_container_size_limit=None,
              page_checksum_verification=False,
-             arrow_extensions_enabled=False):
+             arrow_extensions_enabled=False,
+             read_ree=None):
         """
         Open a parquet file for reading.
 
@@ -1591,6 +1592,7 @@ cdef class ParquetReader(_Weakrefable):
         thrift_container_size_limit : int, optional
         page_checksum_verification : bool, default False
         arrow_extensions_enabled : bool, default False
+        read_ree: iterable[int or str], optional
         """
         cdef:
             shared_ptr[CFileMetaData] c_metadata
@@ -1665,6 +1667,9 @@ cdef class ParquetReader(_Weakrefable):
         if read_dictionary is not None:
             self._set_read_dictionary(read_dictionary, &arrow_props)
 
+        if read_ree is not None:
+            self._set_read_ree(read_ree, &arrow_props)
+
         with nogil:
             check_status(builder.memory_pool(self.pool)
                          .properties(arrow_props)
@@ -1677,6 +1682,13 @@ cdef class ParquetReader(_Weakrefable):
                 column = self.column_name_idx(column)
             props.set_read_dictionary(column, True)
 
+    cdef _set_read_ree(self, read_ree,
+                       ArrowReaderProperties* props):
+        for column in read_ree:
+            if not isinstance(column, int):
+                column = self.column_name_idx(column)
+            props.set_read_ree(column, True)
+
     @property
     def column_paths(self):
         cdef:
diff --git a/python/pyarrow/includes/libarrow_dataset_parquet.pxd b/python/pyarrow/includes/libarrow_dataset_parquet.pxd
index 8f4917f6c0c..7a87e02a958 100644
--- a/python/pyarrow/includes/libarrow_dataset_parquet.pxd
+++ b/python/pyarrow/includes/libarrow_dataset_parquet.pxd
@@ -62,6 +62,7 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
     cdef cppclass CParquetFileFormatReaderOptions \
             "arrow::dataset::ParquetFileFormat::ReaderOptions":
         unordered_set[c_string] dict_columns
+        unordered_set[c_string] ree_columns
         TimeUnit coerce_int96_timestamp_unit
         Type binary_type
         Type list_type
diff --git a/python/pyarrow/includes/libparquet.pxd b/python/pyarrow/includes/libparquet.pxd
index 42d48ba050f..9f2226fe566 100644
--- a/python/pyarrow/includes/libparquet.pxd
+++ b/python/pyarrow/includes/libparquet.pxd
@@ -446,6 +446,8 @@ cdef extern from "parquet/api/reader.h" namespace "parquet" nogil:
         Type list_type()
         void set_read_dictionary(int column_index, c_bool read_dict)
         c_bool read_dictionary(int column_index)
+        void set_read_ree(int column_index, c_bool read_ree)
+        c_bool read_ree(int column_index)
         void set_batch_size(int64_t batch_size)
         int64_t batch_size()
         void set_pre_buffer(c_bool pre_buffer)
diff --git a/python/pyarrow/parquet/core.py b/python/pyarrow/parquet/core.py
index 24cb586c82b..703597763f2 100644
--- a/python/pyarrow/parquet/core.py
+++ b/python/pyarrow/parquet/core.py
@@ -265,6 +265,8 @@ class ParquetFile:
         If True, read Parquet logical types as Arrow extension types where possible,
         (e.g., read JSON as the canonical `arrow.json` extension type or UUID as
         the canonical `arrow.uuid` extension type).
+    read_ree : list
+        List of column names to read directly as an REE Encoded Array.
 
     Examples
     --------
@@ -314,7 +316,7 @@ def __init__(self, source, *, metadata=None, common_metadata=None,
                  coerce_int96_timestamp_unit=None,
                  decryption_properties=None, thrift_string_size_limit=None,
                  thrift_container_size_limit=None, filesystem=None,
-                 page_checksum_verification=False, arrow_extensions_enabled=True):
+                 page_checksum_verification=False, arrow_extensions_enabled=True, read_ree=None):
 
         self._close_source = getattr(source, 'closed', True)
 
@@ -336,6 +338,7 @@ def __init__(self, source, *, metadata=None, common_metadata=None,
             thrift_container_size_limit=thrift_container_size_limit,
             page_checksum_verification=page_checksum_verification,
             arrow_extensions_enabled=arrow_extensions_enabled,
+            read_ree=read_ree,
         )
         self.common_metadata = common_metadata
         self._nested_paths_by_prefix = self._build_nested_paths()
@@ -1350,7 +1353,8 @@ def __init__(self, path_or_paths, filesystem=None, schema=None, *, filters=None,
                  decryption_properties=None, thrift_string_size_limit=None,
                  thrift_container_size_limit=None,
                  page_checksum_verification=False,
-                 arrow_extensions_enabled=True):
+                 arrow_extensions_enabled=True,
+                 read_ree=None):
         import pyarrow.dataset as ds
 
         # map format arguments
@@ -1370,6 +1374,9 @@ def __init__(self, path_or_paths, filesystem=None, schema=None, *, filters=None,
         if read_dictionary is not None:
             read_options.update(dictionary_columns=read_dictionary)
 
+        if read_ree is not None:
+            read_options.update(ree_columns=read_ree)
+
         if decryption_properties is not None:
             read_options.update(decryption_properties=decryption_properties)
 
@@ -1846,7 +1853,8 @@ def read_table(source, *, columns=None, use_threads=True,
                decryption_properties=None, thrift_string_size_limit=None,
                thrift_container_size_limit=None,
                page_checksum_verification=False,
-               arrow_extensions_enabled=True):
+               arrow_extensions_enabled=True,
+               read_ree=None):
 
     try:
         dataset = ParquetDataset(
@@ -1868,6 +1876,7 @@ def read_table(source, *, columns=None, use_threads=True,
             thrift_container_size_limit=thrift_container_size_limit,
             page_checksum_verification=page_checksum_verification,
             arrow_extensions_enabled=arrow_extensions_enabled,
+            read_ree=read_ree
         )
     except ImportError:
         # fall back on ParquetFile for simple cases when pyarrow.dataset
@@ -1902,6 +1911,7 @@ def read_table(source, *, columns=None, use_threads=True,
             thrift_string_size_limit=thrift_string_size_limit,
             thrift_container_size_limit=thrift_container_size_limit,
             page_checksum_verification=page_checksum_verification,
+            read_ree=read_ree
         )
 
     return dataset.read(columns=columns, use_threads=use_threads,
diff --git a/python/pyarrow/tests/parquet/test_basic.py b/python/pyarrow/tests/parquet/test_basic.py
index 591bcffc1ac..66f730bea6a 100644
--- a/python/pyarrow/tests/parquet/test_basic.py
+++ b/python/pyarrow/tests/parquet/test_basic.py
@@ -774,8 +774,9 @@ def test_fastparquet_cross_compatibility(tempdir):
     lambda: pa.array(["", None] * 10).dictionary_encode(),
 ])
 @pytest.mark.parametrize('read_dictionary', [False, True])
+@pytest.mark.parametrize('read_ree', [False, True])
 def test_buffer_contents(
-        array_factory, read_dictionary
+        array_factory, read_dictionary, read_ree
 ):
     # Test that null values are deterministically initialized to zero
     # after a roundtrip through Parquet.
@@ -785,12 +786,18 @@ def test_buffer_contents(
     pq.write_table(orig_table, bio, use_dictionary=True)
     bio.seek(0)
     read_dictionary = ['col'] if read_dictionary else None
+    # TODO: Handle the case where `read_ree` and `read_dictionary` are
+    # both set at the C++ level.
+    read_ree = ['col'] if read_ree and not read_dictionary else None
     table = pq.read_table(bio, use_threads=False,
-                          read_dictionary=read_dictionary)
+                          read_dictionary=read_dictionary,
+                          read_ree=read_ree)
 
     for col in table.columns:
         [chunk] = col.chunks
-        buf = chunk.buffers()[1]
+        # The first two chunks of an REE array aren't related to the nested data,
+        # so use the last chunk.
+        buf = chunk.buffers()[-1]
         assert buf.to_pybytes() == buf.size * b"\0"
 
 
diff --git a/python/pyarrow/tests/parquet/test_data_types.py b/python/pyarrow/tests/parquet/test_data_types.py
index c546bc1532a..c8f1749155f 100644
--- a/python/pyarrow/tests/parquet/test_data_types.py
+++ b/python/pyarrow/tests/parquet/test_data_types.py
@@ -26,6 +26,7 @@
 import pytest
 
 import pyarrow as pa
+import pyarrow.compute as pc
 from pyarrow.tests import util
 from pyarrow.tests.parquet.common import _check_roundtrip, _roundtrip_table
 
@@ -130,7 +131,6 @@ def test_direct_read_dictionary():
 
     data = [
         [util.rands(10) for i in range(nunique)] * repeats,
-
     ]
     table = pa.table(data, names=['f0'])
 
@@ -146,6 +146,29 @@ def test_direct_read_dictionary():
     assert result.equals(expected)
 
 
+
+@pytest.mark.pandas
+def test_direct_read_run_end_encoded():
+    repeats = 10
+    nunique = 5
+
+    data = [
+        [util.rands(10) for i in range(nunique)] * repeats,
+    ]
+    table = pa.table(data, names=['f0'])
+
+    bio = pa.BufferOutputStream()
+    pq.write_table(table, bio)
+    contents = bio.getvalue()
+
+    result = pq.read_table(pa.BufferReader(contents),
+                           read_ree=['f0'])
+
+    # Compute run-end-encoded subfield
+    expected = pa.table([pc.run_end_encode(table[0])], names=['f0'])
+    assert result.equals(expected)
+
+
 @pytest.mark.pandas
 def test_direct_read_dictionary_subfield():
     repeats = 10
diff --git a/python/pyarrow/tests/parquet/test_dataset.py b/python/pyarrow/tests/parquet/test_dataset.py
index d3e9cda7301..077ad35cb89 100644
--- a/python/pyarrow/tests/parquet/test_dataset.py
+++ b/python/pyarrow/tests/parquet/test_dataset.py
@@ -1179,6 +1179,45 @@ def test_dataset_read_dictionary(tempdir):
         assert c0.equals(ex_chunks[1])
         assert c1.equals(ex_chunks[0])
 
+def test_dataset_read_run_end_encoded(tempdir):
+    path = tempdir / "ARROW-32339-dataset"
+    t1 = pa.table([[util.rands(10) for i in range(5)] * 10], names=['f0'])
+    t2 = pa.table([[util.rands(10) for i in range(5)] * 10], names=['f0'])
+    pq.write_to_dataset(t1, root_path=str(path))
+    pq.write_to_dataset(t2, root_path=str(path))
+
+    result = pq.ParquetDataset(
+        path, read_ree=['f0']).read()
+
+    # The order of the chunks is non-deterministic
+    ex_chunks = [pc.run_end_encode(t1[0].chunk(0)),
+                 pc.run_end_encode(t2[0].chunk(0))]
+
+    assert result[0].num_chunks == 2
+    c0, c1 = result[0].chunk(0), result[0].chunk(1)
+    if c0.equals(ex_chunks[0]):
+        assert c1.equals(ex_chunks[1])
+    else:
+        assert c0.equals(ex_chunks[1])
+        assert c1.equals(ex_chunks[0])
+
+def test_dataset_to_batches_run_end_encoded(tempdir):
+    table = pa.table({'a': ["hello"] * 8})
+    filename = tempdir / "test.parquet"
+    pq.write_table(table, filename, row_group_size=2)
+    data_paths = [filename]
+    ree_column = "a"
+    schema = pq.read_schema(filename)
+    filtered_dataset = pa.dataset.FileSystemDataset.from_paths(
+        paths=data_paths,
+        schema=schema,
+        format=pa.dataset.ParquetFileFormat(
+            read_options=pa.dataset.ParquetReadOptions(ree_columns=[ree_column])
+        ),
+        filesystem=pa.fs.LocalFileSystem(),
+    )
+    batch = next(filtered_dataset.to_batches(use_threads=True))
+    assert batch
 
 def test_read_table_schema(tempdir):
     # test that schema keyword is passed through in read_table
diff --git a/python/pyarrow/tests/parquet/test_parquet_file.py b/python/pyarrow/tests/parquet/test_parquet_file.py
index 8eab9bc1740..a549fa04bdf 100644
--- a/python/pyarrow/tests/parquet/test_parquet_file.py
+++ b/python/pyarrow/tests/parquet/test_parquet_file.py
@@ -287,6 +287,16 @@ def test_pre_buffer(pre_buffer):
     assert pf.read().num_rows == N
 
 
+def test_iter_batches_read_ree(tempdir):
+    filename = tempdir / 'pandas_roundtrip.parquet'
+    arrow_table = pa.table({'col1': ["a", "b"], 'col2': [0, 1]})
+    _write_table(arrow_table, filename, version='2.6')
+    ree_col_name = "col1"
+    parquet_file = pq.ParquetFile(filename, read_ree = [ree_col_name])
+    batch = next(parquet_file.iter_batches())
+    assert pa.types.is_run_end_encoded(batch.schema.field(ree_col_name).type)
+
+
 def test_parquet_file_explicitly_closed(tempdir):
     """
     Unopened files should be closed explicitly after use,