GH-45366: [C++][Parquet] Set is_compressed to false when data page v2 is not compressed

cpp/src/parquet/column_writer.cc

@@ -1032,13 +1032,15 @@ void ColumnWriterImpl::BuildDataPageV2(int64_t definition_levels_rle_size,
                                        const std::shared_ptr<Buffer>& values) {
   // Compress the values if needed. Repetition and definition levels are uncompressed in
   // V2.
-  std::shared_ptr<Buffer> compressed_values;
-  if (pager_->has_compressor()) {
+  bool page_is_compressed = false;
+  if (pager_->has_compressor() && values->size() > 0) {
     pager_->Compress(*values, compressor_temp_buffer_.get());
-    compressed_values = compressor_temp_buffer_;
-  } else {
-    compressed_values = values;
+    if (compressor_temp_buffer_->size() < values->size()) {
+      page_is_compressed = true;
+    }
   }
+  std::shared_ptr<Buffer> compressed_values =
+      (page_is_compressed ? compressor_temp_buffer_ : values);
 
   // Concatenate uncompressed levels and the possibly compressed values
   int64_t combined_size =
@@ -1071,14 +1073,14 @@ void ColumnWriterImpl::BuildDataPageV2(int64_t definition_levels_rle_size,
                             combined->CopySlice(0, combined->size(), allocator_));
     std::unique_ptr<DataPage> page_ptr = std::make_unique<DataPageV2>(
         combined, num_values, null_count, num_rows, encoding_, def_levels_byte_length,
-        rep_levels_byte_length, uncompressed_size, pager_->has_compressor(),
+        rep_levels_byte_length, uncompressed_size, page_is_compressed,
         std::move(page_stats), first_row_index, std::move(page_size_stats));
     total_compressed_bytes_ += page_ptr->size() + sizeof(format::PageHeader);
     data_pages_.push_back(std::move(page_ptr));
   } else {
     DataPageV2 page(combined, num_values, null_count, num_rows, encoding_,
                     def_levels_byte_length, rep_levels_byte_length, uncompressed_size,
-                    pager_->has_compressor(), std::move(page_stats), first_row_index,
+                    page_is_compressed, std::move(page_stats), first_row_index,
                     std::move(page_size_stats));
     WriteDataPage(page);
   }

cpp/src/parquet/column_writer_test.cc

@@ -410,7 +410,7 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
 
   const ColumnDescriptor* descr_;
 
- private:
+ protected:
   std::unique_ptr<ColumnChunkMetaDataBuilder> metadata_;
   std::shared_ptr<::arrow::io::BufferOutputStream> sink_;
   std::shared_ptr<WriterProperties> writer_properties_;
@@ -1807,5 +1807,50 @@ TEST_F(TestValuesWriterInt32Type, AllNullsCompressionInPageV2) {
   }
 }
 
+#ifdef ARROW_WITH_ZSTD
+
+TEST_F(TestValuesWriterInt32Type, AvoidCompressedInDataPageV2) {
+  Compression::type compression = Compression::ZSTD;
+  auto verify_only_one_uncompressed_page = [&](int total_num_values) {
+    ColumnProperties column_properties;
+    column_properties.set_compression(compression);
+
+    auto writer =
+        this->BuildWriter(SMALL_SIZE, column_properties, ParquetVersion::PARQUET_2_LATEST,
+                          ParquetDataPageVersion::V2);
+    writer->WriteBatch(static_cast<int64_t>(values_.size()), this->def_levels_.data(),
+                       nullptr, this->values_ptr_);
+    writer->Close();
+    ASSERT_OK_AND_ASSIGN(auto buffer, this->sink_->Finish());
+    auto source = std::make_shared<::arrow::io::BufferReader>(buffer);
+    ReaderProperties readerProperties;
+    std::unique_ptr<PageReader> page_reader = PageReader::Open(
+        std::move(source), total_num_values, compression, readerProperties);
+    auto data_page = std::static_pointer_cast<DataPageV2>(page_reader->NextPage());
+    ASSERT_TRUE(data_page != nullptr);
+    ASSERT_FALSE(data_page->is_compressed());
+    ASSERT_TRUE(page_reader->NextPage() == nullptr);
+  };
+  {
+    // zero-sized data buffer should be handled correctly.
+    this->SetUpSchema(Repetition::OPTIONAL);
+    this->GenerateData(SMALL_SIZE);
+    std::fill(this->def_levels_.begin(), this->def_levels_.end(), 0);
+    verify_only_one_uncompressed_page(SMALL_SIZE);
+  }
+  {
+    // When only compress little data, the compressed size would even be
+    // larger than the original size. In this case, the `is_compressed` flag
+    // should be set to false.
+    this->SetUpSchema(Repetition::OPTIONAL);
+    this->GenerateData(1);
+    std::fill(this->def_levels_.begin(), this->def_levels_.end(), 1);
+    values_[0] = 142857;
+    verify_only_one_uncompressed_page(/*total_num_values=*/1);
+  }
+}
+
+#endif
+
 }  // namespace test
 }  // namespace parquet