ARROW-13330: [Go][Parquet] Add the rest of the Encoding package

go/parquet/internal/encoding/boolean_encoder.go

@@ -74,11 +74,11 @@ func (enc *PlainBooleanEncoder) EstimatedDataEncodedSize() int64 {
 
 // FlushValues returns the buffered data, the responsibility is on the caller
 // to release the buffer memory
-func (enc *PlainBooleanEncoder) FlushValues() Buffer {
+func (enc *PlainBooleanEncoder) FlushValues() (Buffer, error) {
 	if enc.wr.Pos() > 0 {
 		toFlush := int(enc.wr.Pos())
 		enc.append(enc.bitsBuffer[:bitutil.BytesForBits(int64(toFlush))])
 	}
 
-	return enc.sink.Finish()
+	return enc.sink.Finish(), nil
 }

go/parquet/internal/encoding/delta_bit_packing.go

@@ -428,7 +428,8 @@ func (enc *deltaBitPackEncoder) putInternal(data interface{}) {
 }
 
 // FlushValues flushes any remaining data and returns the finished encoded buffer
-func (enc *deltaBitPackEncoder) FlushValues() Buffer {
+// or returns nil and any error encountered during flushing.
+func (enc *deltaBitPackEncoder) FlushValues() (Buffer, error) {
 	if enc.bitWriter != nil {
 		// write any remaining values
 		enc.flushBlock()
@@ -457,7 +458,7 @@ func (enc *deltaBitPackEncoder) FlushValues() Buffer {
 
 		buffer = append(buffer, flushed.Buf()[:enc.bitWriter.Written()]...)
 	}
-	return poolBuffer{memory.NewBufferBytes(buffer)}
+	return poolBuffer{memory.NewBufferBytes(buffer)}, nil
 }
 
 // EstimatedDataEncodedSize returns the current amount of data actually flushed out and written

go/parquet/internal/encoding/delta_byte_array.go

@@ -105,21 +105,28 @@ func (enc *DeltaByteArrayEncoder) PutSpaced(in []parquet.ByteArray, validBits []
 }
 
 // Flush flushes any remaining data out and returns the finished encoded buffer.
-func (enc *DeltaByteArrayEncoder) FlushValues() Buffer {
+// or returns nil and any error encountered during flushing.
+func (enc *DeltaByteArrayEncoder) FlushValues() (Buffer, error) {
 	if enc.prefixEncoder == nil {
 		enc.initEncoders()
 	}
-	prefixBuf := enc.prefixEncoder.FlushValues()
+	prefixBuf, err := enc.prefixEncoder.FlushValues()
+	if err != nil {
+		return nil, err
+	}
 	defer prefixBuf.Release()
 
-	suffixBuf := enc.suffixEncoder.FlushValues()
+	suffixBuf, err := enc.suffixEncoder.FlushValues()
+	if err != nil {
+		return nil, err
+	}
 	defer suffixBuf.Release()
 
 	ret := bufferPool.Get().(*memory.Buffer)
 	ret.ResizeNoShrink(prefixBuf.Len() + suffixBuf.Len())
 	copy(ret.Bytes(), prefixBuf.Bytes())
 	copy(ret.Bytes()[prefixBuf.Len():], suffixBuf.Bytes())
-	return poolBuffer{ret}
+	return poolBuffer{ret}, nil
 }
 
 // DeltaByteArrayDecoder is a decoder for a column of data encoded using incremental or prefix encoding.

go/parquet/internal/encoding/delta_length_byte_array.go

@@ -71,9 +71,13 @@ func (DeltaLengthByteArrayEncoder) Type() parquet.Type {
 	return parquet.Types.ByteArray
 }
 
-// FlushValues flushes any remaining data and returns the final encoded buffer of data.
-func (enc *DeltaLengthByteArrayEncoder) FlushValues() Buffer {
-	ret := enc.lengthEncoder.FlushValues()
+// FlushValues flushes any remaining data and returns the final encoded buffer of data
+// or returns nil and any error encountered.
+func (enc *DeltaLengthByteArrayEncoder) FlushValues() (Buffer, error) {
+	ret, err := enc.lengthEncoder.FlushValues()
+	if err != nil {
+		return nil, err
+	}
 	defer ret.Release()
 
 	data := enc.sink.Finish()
@@ -83,7 +87,7 @@ func (enc *DeltaLengthByteArrayEncoder) FlushValues() Buffer {
 	output.ResizeNoShrink(ret.Len() + data.Len())
 	copy(output.Bytes(), ret.Bytes())
 	copy(output.Bytes()[ret.Len():], data.Bytes())
-	return poolBuffer{output}
+	return poolBuffer{output}, nil
 }
 
 // DeltaLengthByteArrayDecoder is a decoder for handling data produced by the corresponding

go/parquet/internal/encoding/encoder.go

@@ -99,7 +99,7 @@ func (e *encoder) append(data []byte)              { e.sink.Write(data) }
 // FlushValues flushes any unwritten data to the buffer and returns the finished encoded buffer of data.
 // This also clears the encoder, ownership of the data belongs to whomever called FlushValues, Release
 // should be called on the resulting Buffer when done.
-func (e *encoder) FlushValues() Buffer { return e.sink.Finish() }
+func (e *encoder) FlushValues() (Buffer, error) { return e.sink.Finish(), nil }
 
 // Bytes returns the current bytes that have been written to the encoder's buffer but doesn't transfer ownership.
 func (e *encoder) Bytes() []byte { return e.sink.Bytes() }
@@ -147,13 +147,17 @@ func (d *dictEncoder) addIndex(idx int) {
 }
 
 // FlushValues dumps all the currently buffered indexes that would become the data page to a buffer and
-// returns it.
-func (d *dictEncoder) FlushValues() Buffer {
+// returns it or returns nil and any error encountered.
+func (d *dictEncoder) FlushValues() (Buffer, error) {
 	buf := bufferPool.Get().(*memory.Buffer)
 	buf.Reserve(int(d.EstimatedDataEncodedSize()))
-	size := d.WriteIndices(buf.Buf())
+	size, err := d.WriteIndices(buf.Buf())
+	if err != nil {
+		poolBuffer{buf}.Release()
+		return nil, err
+	}
 	buf.ResizeNoShrink(size)
-	return poolBuffer{buf}
+	return poolBuffer{buf}, nil
 }
 
 // EstimatedDataEncodedSize returns the maximum number of bytes needed to store the RLE encoded indexes, not including the
@@ -187,19 +191,20 @@ func (d *dictEncoder) WriteDict(out []byte) {
 
 // WriteIndices performs Run Length encoding on the indexes and the writes the encoded
 // index value data to the provided byte slice, returning the number of bytes actually written.
-func (d *dictEncoder) WriteIndices(out []byte) int {
+// If any error is encountered, it will return -1 and the error.
+func (d *dictEncoder) WriteIndices(out []byte) (int, error) {
 	out[0] = byte(d.BitWidth())
 
 	enc := utils.NewRleEncoder(utils.NewWriterAtBuffer(out[1:]), d.BitWidth())
 	for _, idx := range d.idxValues {
-		if !enc.Put(uint64(idx)) {
-			return -1
+		if err := enc.Put(uint64(idx)); err != nil {
+			return -1, err
 		}
 	}
 	nbytes := enc.Flush()
 
 	d.idxValues = d.idxValues[:0]
-	return nbytes + 1
+	return nbytes + 1, nil
 }
 
 // Put adds a value to the dictionary data column, inserting the value if it