PARQUET-701: Ensure that Close can be called multiple times

Author: Uwe L. Korn <uwelk@xhochy.com>

Closes apache#149 from xhochy/parquet-701 and squashes the following commits:

6f29d8b [Uwe L. Korn] PARQUET-701: Ensure that Close can be called multiple times

Change-Id: I55fc9bc431365b7b7665a1e065d139b5a8b18651

cpp/src/parquet/column/column-writer-test.cc

@@ -109,6 +109,8 @@ class TestPrimitiveWriter : public ::testing::Test {
  std::unique_ptr<TypedColumnWriter<TestType>> writer =
  this->BuildWriter(SMALL_SIZE, encoding);
  writer->WriteBatch(this->values_.size(), nullptr, nullptr, this->values_ptr_);
+ // The behaviour should be independent from the number of Close() calls
+ writer->Close();
  writer->Close();
 
  this->ReadColumn();

cpp/src/parquet/column/writer.cc

@@ -46,7 +46,8 @@ ColumnWriter::ColumnWriter(const ColumnDescriptor* descr,
  num_buffered_values_(0),
  num_buffered_encoded_values_(0),
  num_rows_(0),
- total_bytes_written_(0) {
+ total_bytes_written_(0),
+ closed_(false) {
  InitSinks();
 }
 
@@ -56,11 +57,13 @@ void ColumnWriter::InitSinks() {
 }
 
 void ColumnWriter::WriteDefinitionLevels(int64_t num_levels, const int16_t* levels) {
+ DCHECK(!closed_);
  definition_levels_sink_->Write(
  reinterpret_cast<const uint8_t*>(levels), sizeof(int16_t) * num_levels);
 }
 
 void ColumnWriter::WriteRepetitionLevels(int64_t num_levels, const int16_t* levels) {
+ DCHECK(!closed_);
  repetition_levels_sink_->Write(
  reinterpret_cast<const uint8_t*>(levels), sizeof(int16_t) * num_levels);
 }
@@ -129,12 +132,15 @@ void ColumnWriter::WriteDataPage(const DataPage& page) {
 }
 
 int64_t ColumnWriter::Close() {
- if (has_dictionary_) { WriteDictionaryPage(); }
- // Write all outstanding data to a new page
- if (num_buffered_values_ > 0) { AddDataPage(); }
-
- for (size_t i = 0; i < data_pages_.size(); i++) {
- WriteDataPage(data_pages_[i]);
+ if (!closed_) {
+ closed_ = true;
+ if (has_dictionary_) { WriteDictionaryPage(); }
+ // Write all outstanding data to a new page
+ if (num_buffered_values_ > 0) { AddDataPage(); }
+
+ for (size_t i = 0; i < data_pages_.size(); i++) {
+ WriteDataPage(data_pages_[i]);
+ }
  }
 
  if (num_rows_ != expected_rows_) {

cpp/src/parquet/column/writer.h

@@ -101,6 +101,7 @@ class PARQUET_EXPORT ColumnWriter {
  int num_rows_;
 
  int total_bytes_written_;
+ bool closed_;
 
  std::unique_ptr<InMemoryOutputStream> definition_levels_sink_;
  std::unique_ptr<InMemoryOutputStream> repetition_levels_sink_;