57 alltypes plain writer (tensorflow#65)

* started working on boolean writer

* - add plain writer for float, int64 and double
- dont' read definitions for required columns

* bitpacker for bolean

* writing strings

* test for reading/writing bools

* finish reading basic types

* update readme

Author: Ivan Gavryliuk

PULL_REQUEST_TEMPLATE.md

@@ -1,7 +1,10 @@
-Fixes #
+### Fixes
+
+Issue #
 
 ### Description
 
+desctiption goes here
 
 ### Todos
 - [ ] unit/integration tests

README.md

@@ -12,7 +12,7 @@ This project is aimed to fix this problem.
 
 ## Roadmap
 
-We have just started to work on this library, contributors are welcome.
+We have just started to work on this library, [contributors are welcome](CONTRIBUTING.md).
 
 |Phase|Description|State|
 |-----|-----------|-----|
@@ -21,13 +21,15 @@ We have just started to work on this library, contributors are welcome.
 |3|Support GZIP and SNAPPY decompression/compression|planning|
 |4|Integrate with popular products like Azure Data Lakes|planning|
 
+You can track the amount of features we have [implemented so far](doc/features.md).
+
 ## Getting started
 
 **parquet-dotnet** is redistributed as a [NuGet package](https://www.nuget.org/packages/Parquet.Net) for `.NET 4.5.1` and `.NET Standard 1.6`. All code is managed and doesn't have any native dependencies, therefore you are ready to go after referencing the package.
 
 ### Reading files
 
-In order to read a parquet file you need to open a stream first. Due to the fact that Parquet utilises file seeking extensively, the input stream must be *readable and seekable*.
+In order to read a parquet file you need to open a stream first. Due to the fact that Parquet utilises file seeking extensively, the input stream must be *readable and seekable*. This somewhat limits the amount of streaming you can do, for instance you can't read a parquet file from a network stream as we need to jump around it, therefore you have to download it locally to disk and then open.
 
 For instance, to read a file `c:\test.parquet` you woudl normally write the following code
 
@@ -46,6 +48,33 @@ using(Stream fs = File.OpenRead("c:\\test.parquet"))
 
 this will read entire file in memory as a set of columns inside `ParquetDataSet` class.
 
+### Writing files
+
+Parquet.Net operates on streams, therefore you need to create it first. The following example shows how to create a file on disk with two columns - `id` and `city`.
+
+```csharp
+using System.IO;
+using Parquet;
+
+var idColumn = new ParquetColumn<int>("id");
+idColumn.Add(1, 2);
+
+var cityColumn = new ParquetColumn<string>("city");
+cityColumn.Add("London", "Derby");
+
+var dataSet = new ParquetDataSet(idColumn, cityColumn);
+
+using(Stream fileStream = File.OpenWrite("c:\\test.parquet"))
+{
+	using(var writer = new ParquetWriter(fileStream))
+	{
+		writer.Write(dataSet);
+	}
+}
+
+```
+
+
 ## Tools
 
 ### parq

src/Parquet.Test/File/Values/PlainValuesReaderWriterTest.cs

@@ -0,0 +1,52 @@
+using Parquet.File.Values;
+using Parquet.Thrift;
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Text;
+using Xunit;
+
+namespace Parquet.Test.File.Values
+{
+   public class PlainValuesReaderWriterTest
+   {
+      private ParquetOptions _options;
+      private IValuesReader _reader;
+      private IValuesWriter _writer;
+      private BinaryWriter _bw;
+      private BinaryReader _br;
+      private MemoryStream _ms;
+
+      public PlainValuesReaderWriterTest()
+      {
+         _options = new ParquetOptions();
+
+         _reader = new PlainValuesReader(_options);
+         _writer = new PlainValuesWriter(_options);
+
+         _ms = new MemoryStream();
+         _br = new BinaryReader(_ms);
+         _bw = new BinaryWriter(_ms);
+      }
+
+      [Fact]
+      public void Booleans()
+      {
+         var bools = new List<bool>();
+         bools.AddRange(new[] { true, false, true });
+
+         var schema = new SchemaElement();
+         schema.Type = Thrift.Type.BOOLEAN;
+
+         _writer.Write(_bw, schema, bools);
+
+         _ms.Position = 0;
+
+         var boolsRead = new List<bool?>();
+         _reader.Read(_br, schema, boolsRead, 3);
+
+         Assert.Equal(bools, boolsRead.Cast<bool>().ToList());
+      }
+   }
+}

src/Parquet.Test/Parquet.Test.csproj

@@ -106,4 +106,8 @@
     <Service Include="{82a7f48d-3b50-4b1e-b82e-3ada8210c358}" />
   </ItemGroup>
 
+  <ItemGroup>
+    <Folder Include="Writer\" />
+  </ItemGroup>
+
 </Project>

src/Parquet.Test/ParquetReaderOnTestFilesTest.cs

@@ -5,7 +5,7 @@
 
 namespace Parquet.Test
 {
-   using File = System.IO.File;
+   using F = System.IO.File;
 
    /// <summary>
    /// Tests a set of predefined test files that they read back correct
@@ -15,7 +15,7 @@ public class ParquetReaderOnTestFilesTest
       [Fact]
       public void FixedLenByteArray_dictionary()
       {
-         using (Stream s = File.OpenRead(GetDataFilePath("fixedlenbytearray.parquet")))
+         using (Stream s = F.OpenRead(GetDataFilePath("fixedlenbytearray.parquet")))
          {
             using (var r = new ParquetReader(s))
             {
@@ -27,7 +27,7 @@ public void FixedLenByteArray_dictionary()
       [Fact]
       public void Datetypes_all()
       {
-         using (Stream s = File.OpenRead(GetDataFilePath("dates.parquet")))
+         using (Stream s = F.OpenRead(GetDataFilePath("dates.parquet")))
          {
             using (var r = new ParquetReader(s))
             {

src/Parquet.Test/ParquetWriterTest.cs

@@ -6,32 +6,40 @@ namespace Parquet.Test
 {
    public class ParquetWriterTest
    {
-      //[Fact]
+      [Fact]
       public void Write_simple_int32_and_int_reads_back()
       {
-         const string path = "c:\\tmp\\first.parquet";
-         if (F.Exists(path)) F.Delete(path);
+         var ms = new MemoryStream();
 
-         using (var ms = F.OpenWrite(path))
+         using (var writer = new ParquetWriter(ms))
          {
-            using (var writer = new ParquetWriter(ms))
-            {
-               var ds = new ParquetDataSet();
+            var ds = new ParquetDataSet();
+
+            //8 values for each column
+
+            var idCol = new ParquetColumn<int>("id");
+            idCol.Add(4, 5, 6, 7, 2, 3, 0, 1);
+
+            var bool_col = new ParquetColumn<bool>("bool_col");
+            bool_col.Add(true, false, true, false, true, false, true, false);
 
-               var intCol = new ParquetColumn<int>("id");
-               intCol.Add(1, 2, 3, 4, 5);
+            var string_col = new ParquetColumn<string>("string_col");
+            string_col.Add("0", "1", "0", "1", "0", "1", "0", "1");
 
-               writer.Write(new ParquetDataSet(intCol));
-            }
+            writer.Write(new ParquetDataSet(idCol, bool_col, string_col));
          }
 
-         using (var ms = F.OpenRead(path))
+         ms.Position = 0;
+         using (var reader = new ParquetReader(ms))
          {
-            using (var reader = new ParquetReader(ms))
-            {
-               ParquetDataSet ds = reader.Read();
-            }
+            ParquetDataSet ds = reader.Read();
          }
+
+#if DEBUG
+         const string path = "c:\\tmp\\first.parquet";
+         F.WriteAllBytes(path, ms.ToArray());
+#endif
+
       }
    }
 }

src/Parquet.Test/Reader/ParquetCsvComparison.cs

@@ -42,12 +42,6 @@ private void Compare(ParquetDataSet parquet, Dictionary<string, List<string>> cs
          int i = 0;
          foreach(ParquetColumn pc in parquet.Columns)
          {
-            /*if (pc.Name == "Parish")
-            {
-               i += 1;
-               continue;
-            }*/
-
             List<string> cc = csv[pc.Name];
             Type expectedColumnType = columnTypes[i];
 

src/Parquet.sln

@@ -23,7 +23,9 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
 		..\appveyor.ps1 = ..\appveyor.ps1
 		..\build.ps1 = ..\build.ps1
 		..\CONTRIBUTING.md = ..\CONTRIBUTING.md
+		..\ISSUE_TEMPLATE.md = ..\ISSUE_TEMPLATE.md
 		PreCommit.ps1 = PreCommit.ps1
+		..\PULL_REQUEST_TEMPLATE.md = ..\PULL_REQUEST_TEMPLATE.md
 		..\README.md = ..\README.md
 	EndProjectSection
 EndProject

src/Parquet/CompressionMethod.cs

@@ -0,0 +1,13 @@
+namespace Parquet
+{
+   /// <summary>
+   /// Parquet compression method
+   /// </summary>
+   public enum CompressionMethod
+   {
+      /// <summary>
+      /// No compression
+      /// </summary>
+      None
+   }
+}

src/Parquet/File/Data/GzipDataWriter.cs

@@ -0,0 +1,12 @@
+using System;
+using System.IO.Compression;
+
+namespace Parquet.File.Data
+{
+   class GzipDataWriter : IDataWriter
+   {
+      public void Write(byte[] buffer)
+      {
+      }
+   }
+}

src/Parquet/File/Data/UncompressedDataWriter.cs

@@ -1,14 +1,19 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
+using System.IO;
 
 namespace Parquet.File.Data
 {
    class UncompressedDataWriter : IDataWriter
    {
+      private readonly Stream _stream;
+
+      public UncompressedDataWriter(Stream stream)
+      {
+         _stream = stream;
+      }
+
       public void Write(byte[] buffer)
       {
-         throw new NotImplementedException();
+         _stream.Write(buffer, 0, buffer.Length);
       }
    }
 }

src/Parquet/File/ListFactory.cs

@@ -0,0 +1,47 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using Type = System.Type;
+using TType = Parquet.Thrift.Type;
+using System.Collections;
+using Parquet.Thrift;
+
+namespace Parquet.File
+{
+   static class ListFactory
+   {
+      struct TypeTag
+      {
+         public TType PType;
+
+         public Func<IList> Create;
+
+         public ConvertedType? ConvertedType;
+
+         public TypeTag(TType ptype, Func<IList> create, ConvertedType? convertedType)
+         {
+            PType = ptype;
+            Create = create;
+            ConvertedType = convertedType;
+         }
+      }
+
+      private static readonly Dictionary<Type, TypeTag> TypeToTag = new Dictionary<Type, TypeTag>
+      {
+         { typeof(int), new TypeTag(TType.INT32, () => new List<int>(), null) },
+         { typeof(bool), new TypeTag(TType.BOOLEAN, () => new List<bool>(), null) },
+         { typeof(string), new TypeTag(TType.BYTE_ARRAY, () => new List<string>(), ConvertedType.UTF8) }
+      };
+
+      public static IList Create(Type systemType, SchemaElement schema)
+      {
+         if (!TypeToTag.TryGetValue(systemType, out TypeTag tag))
+            throw new NotSupportedException($"system type {systemType} is not supported");
+
+         schema.Type = tag.PType;
+         if(tag.ConvertedType != null)
+            schema.Converted_type = tag.ConvertedType.Value;
+         return tag.Create();
+      }
+   }
+}

src/Parquet/File/MetaBuilder.cs

@@ -0,0 +1,40 @@
+using Parquet.Thrift;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+
+namespace Parquet.File
+{
+   class MetaBuilder
+   {
+      private readonly FileMetaData _meta;
+
+      public MetaBuilder()
+      {
+         _meta = new FileMetaData();
+
+         _meta.Created_by = "parquet-dotnet";
+         _meta.Version = 1;
+         _meta.Row_groups = new List<RowGroup>();
+      }
+
+      public FileMetaData ThriftMeta => _meta;
+
+      public void AddSchema(ParquetDataSet dataSet)
+      {
+         long totalCount = dataSet.Count;
+
+         _meta.Schema = new List<SchemaElement> { new SchemaElement("schema") { Num_children = dataSet.Columns.Count } };
+         _meta.Schema.AddRange(dataSet.Columns.Select(c => c.Schema));
+         _meta.Num_rows = totalCount;
+      }
+
+      public RowGroup AddRowGroup()
+      {
+         var rg = new RowGroup();
+         _meta.Row_groups.Add(rg);
+         return rg;
+      }
+   }
+}

src/Parquet/File/PColumn.cs

@@ -140,7 +140,11 @@ private IList ReadDictionaryPage(PageHeader ph)
             {
                //todo: read repetition levels (only relevant for nested columns)
 
-               List<int> definitions = ReadDefinitionLevels(reader, (int)maxValues);
+               //check if there are definitions at all
+               bool hasDefinitions = _schemaElement.Repetition_type == FieldRepetitionType.OPTIONAL;
+               List<int> definitions = hasDefinitions
+                  ? ReadDefinitionLevels(reader, (int)maxValues)
+                  : null;
 
                // these are pointers back to the Values table - lookup on values 
                List<int> indexes = ReadColumnValues(reader, ph.Data_page_header.Encoding, destination, maxValues);