Allow parallel encoding of multiple columns of single parquet file.

[avakaluk]

Hi,
I recently found following discussion apache/arrow-rs#1718 about allowing parallel writes to same parquet file/row group. Are there any plans to have it in ParquetSharp?
Alex

[adamreeve]

Hi, that discussion is for the Rust Parquet library, but ParquetSharp uses the C++ library internally. The Parquet C++ library and therefore ParquetSharp does support multi-threaded writing although it's not something I've experimented much with. If you're writing Arrow data, you can enable multi-threaded writing in the ArrowWriterProperties.

Otherwise you'll need to write in buffered mode, using AppendBuffferedRowGroup. Then you can get column writers that can be written to concurrently with the Column(int) method.

[avakaluk]

Thank you, for your prompt reply. Will try it.

[avakaluk]

Hi,
I tested working with BufferedRowGroup to perform multi-threaded write and it generates corrupted parquet file. This is the error that I get when I try to read second column:

Message: 
Test method Teleduct.PostProcessingTests.ColumnDataReadWriteTest.SampleTestAsync threw exception:
System.ArgumentException: read 0 rows, expected 100 rows

Stack Trace: 
LogicalColumnReader`1.ReadAll(Int32 rows)

This is sample code that I run to reproduce errors:

   using (var stream = new MemoryStream())
   {
       var columns = new Column[]
       {
               new Column<string>("Column1"),
               new Column<int?>("Column2")
       };

       using (var writer = new ParquetFileWriter(new ManagedOutputStream(stream, true), columns))
       {
           using (var rowGroup = writer.AppendBufferedRowGroup())
           {
               var tasks = writer.Schema.GroupNode.Fields.Select((field, index) =>
               {
                   if (field.LogicalType.Type == LogicalTypeEnum.String)
                   {
                       return Task.Run(() => WriteStringData(rowGroup.Column(index).LogicalWriter<string?>()));
                   }
                   else if (field.LogicalType.Type == LogicalTypeEnum.Int)
                   {
                       return Task.Run(() => WrtiteIntData(rowGroup.Column(index).LogicalWriter<int?>()));
                   }
                   else
                   {
                       throw new InvalidOperationException("Unsupported type");
                   }
               }).ToArray();

               await Task.WhenAll(tasks);

               rowGroup.Close();
           }
           writer.Close();
       }

       stream.Position = 0;
       using var reader = new ParquetFileReader(stream);
       for (var rg = 0; rg < reader.FileMetaData.NumRowGroups; rg++)
       {
           using var rowGroup = reader.RowGroup(rg);
           using var col = rowGroup.Column(0).LogicalReader<string?>();
           var values = col.ReadAll((int)rowGroup.MetaData.NumRows);
           foreach (var value in values)
           {
               Console.WriteLine(value);
           }

           using var col2 = rowGroup.Column(1).LogicalReader<int?>();
           var values2 = col.ReadAll((int)rowGroup.MetaData.NumRows);
           foreach (var value in values2)
           {
               Console.WriteLine(value);
           }
       }

   }

[adamreeve]

You have an error on this line:

var values2 = col.ReadAll((int)rowGroup.MetaData.NumRows);

You are reading from col instead of col2. This should be:

var values2 = col2.ReadAll((int)rowGroup.MetaData.NumRows);

If I change that then your example works for me.

[avakaluk]

You are right. I need to work on my reproduce scenario to better present error I have, once I write using ColumnWriter to BufferedRowGroup from multiple threads. Will try to work on this. This is error I get, while reading last column:
ParquetSharp.ParquetException: class parquet::ParquetException (message: 'Unexpected end of stream')