Test for ParquetWriter

Author: caetera

ThermoRawFileParserTest/ThermoRawFileParserTest.csproj

@@ -32,6 +32,7 @@
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
+    <PackageReference Include="Parquet.Net" Version="5.0.1" />
   </ItemGroup>
 
   <ItemGroup>

ThermoRawFileParserTest/WriterTests.cs

@@ -5,6 +5,7 @@
 using System.Xml.Serialization;
 using IO.Mgf;
 using NUnit.Framework;
+using Parquet;
 using ThermoRawFileParser;
 using ThermoRawFileParser.Writer.MzML;
 
@@ -283,7 +284,7 @@ public void TestMzML_MS2()
         }
 
         [Test]
-        public void TestParquet()
+        public void TestParquetCentroid()
         {
             // Get temp path for writing the test mzML
             var tempFilePath = Path.GetTempPath();
@@ -294,17 +295,45 @@ public void TestParquet()
             RawFileParser.Parse(parseInput);
 
             // Actual test
-            //var xmlSerializer = new XmlSerializer(typeof(mzMLType));
-            //var testMzMl = (mzMLType)xmlSerializer.Deserialize(new FileStream(
-            //    Path.Combine(tempFilePath, "small.mzML"), FileMode.Open, FileAccess.Read, FileShare.ReadWrite));
+            var parquetFilePath = Path.Combine(tempFilePath, "small.mzparquet");
 
-            //Assert.That(testMzMl.run.spectrumList.count, Is.EqualTo("48"));
-            //Assert.That(testMzMl.run.spectrumList.spectrum.Length, Is.EqualTo(48));
+            using (var parquetReader = ParquetReader.CreateAsync(parquetFilePath).Result)
+            {
+                var groupReader = parquetReader.OpenRowGroupReader(0);
+                var schema = parquetReader.Schema;
+                var scanColumn = groupReader.ReadColumnAsync(schema.FindDataField("scan")).Result;
 
-            //Assert.That(testMzMl.run.chromatogramList.count, Is.EqualTo("1"));
-            //Assert.That(testMzMl.run.chromatogramList.chromatogram.Length, Is.EqualTo(1));
+                Assert.That(scanColumn.NumValues, Is.EqualTo(48520));
+                Assert.That(scanColumn.Statistics.DistinctCount, Is.EqualTo(48));
+                Assert.That((from int p in scanColumn.Data where p == 22 select p).Count(), Is.EqualTo(1632));
+            }
+        }
+
+        [Test]
+        public void TestParquetProfile()
+        {
+            // Get temp path for writing the test mzML
+            var tempFilePath = Path.GetTempPath();
 
-            //Assert.That(testMzMl.run.chromatogramList.chromatogram[0].defaultArrayLength, Is.EqualTo(48));
+            var testRawFile = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, @"Data/small.RAW");
+            var parseInput = new ParseInput(testRawFile, null, tempFilePath, OutputFormat.Parquet);
+            parseInput.NoPeakPicking = new HashSet<int> { 1, 2 };
+
+            RawFileParser.Parse(parseInput);
+
+            // Actual test
+            var parquetFilePath = Path.Combine(tempFilePath, "small.mzparquet");
+
+            using (var parquetReader = ParquetReader.CreateAsync(parquetFilePath).Result)
+            {
+                var groupReader = parquetReader.OpenRowGroupReader(0);
+                var schema = parquetReader.Schema;
+                var scanColumn = groupReader.ReadColumnAsync(schema.FindDataField("scan")).Result;
+
+                Assert.That(scanColumn.NumValues, Is.EqualTo(305213));
+                Assert.That(scanColumn.Statistics.DistinctCount, Is.EqualTo(48));
+                Assert.That((from int p in scanColumn.Data where p == 22 select p).Count(), Is.EqualTo(17758));
+            }
         }
     }
 }

Writer/MzMlSpectrumWriter.cs

@@ -641,21 +641,6 @@ public override void Write(IRawDataPlus rawFile, int firstScanNumber, int lastSc
 
                 Writer.Flush();
                 Writer.Close();
-
-                //This section is not necessary?
-                /*if (_doIndexing)
-                {
-                    try
-                    {
-                        cryptoStream.Flush();
-                        cryptoStream.Close();
-                    }
-                    catch (System.ObjectDisposedException e)
-                    {
-                        // Cannot access a closed file.  CryptoStream was already closed when closing _writer
-                        Log.Warn($"Warning: {e.Message}");
-                    }
-                }*/
             }
 
             // In case of indexed mzML, change the extension from xml to mzML and check for the gzip option

Writer/ParquetSpectrumWriter.cs

@@ -36,6 +36,8 @@ public class ParquetSpectrumWriter : SpectrumWriter
         private static readonly ILog Log =
             LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
 
+        private const int ParquetRowGroupSize = 1_048_576;
+
         public ParquetSpectrumWriter(ParseInput parseInput) : base(parseInput)
         {
             //nothing to do here
@@ -49,7 +51,7 @@ public override void Write(IRawDataPlus raw, int firstScanNumber, int lastScanNu
             }
 
             ConfigureWriter(".mzparquet");
-            
+
             ParquetSerializerOptions opts = new ParquetSerializerOptions();
             opts.CompressionLevel = System.IO.Compression.CompressionLevel.Fastest;
             opts.CompressionMethod = Parquet.CompressionMethod.Zstd;
@@ -91,7 +93,7 @@ public override void Write(IRawDataPlus raw, int firstScanNumber, int lastScanNu
                 // - some row groups might have more than this number of ions
                 //   but this ensures that all ions from a single scan are always
                 //   present in the same row group (critical property of mzparquet)
-                if (data.Count >= 1_048_576)
+                if (data.Count >= ParquetRowGroupSize)
                 {
                     var task = ParquetSerializer.SerializeAsync(data, Writer.BaseStream, opts);
                     task.Wait();
@@ -108,6 +110,10 @@ public override void Write(IRawDataPlus raw, int firstScanNumber, int lastScanNu
                 task.Wait();
                 Log.Debug("Writing final row group");
             }
+
+            // Release the OS file handle
+            Writer.Flush();
+            Writer.Close();
         }
 
         private void AddScan(IRawDataPlus raw, int scanNumber, List<MzParquet> data)