Spark compatibility adjustments

Author: Ivan Gavryliuk

src/Parquet.Test/MapsTest.cs

@@ -18,14 +18,15 @@ public void Simple_first_level_map_int_to_string()
          ds.Add(1, new Dictionary<int, string>
          {
             [1] = "one",
-            [2] = "two"
+            [2] = "two",
+            [3] = "three"
          });
 
          //ParquetWriter.WriteFile(ds, "c:\\tmp\\map.parquet");
 
          DataSet ds1 = DataSetGenerator.WriteRead(ds);
 
-         Assert.Equal("{1;[1=>one;2=>two]}", ds1[0].ToString());
+         Assert.Equal("{1;[1=>one;2=>two;3=>three]}", ds1[0].ToString());
       }
    }
 }

src/Parquet/File/ColumnWriter.cs

@@ -80,7 +80,7 @@ private List<PageTag> WriteValues(SchemaElement schema, IList values, Thrift.Pag
             ph.Data_page_header.Num_values = values.Count;
          }
 
-         if (schema.IsNullable || schema.MaxRepetitionLevel > 0)
+         if (schema.IsNullable || schema.MaxDefinitionLevel > 0)
          {
             var dpack = new DefinitionPack(_schema, _formatOptions);
             values = dpack.Unpack(values, out definitions);

src/Parquet/File/FileMetadataBuilder.cs

@@ -101,7 +101,8 @@ private static void AddMapSchema(List<TSchemaElement> container, SchemaElement s
          var root = new TSchemaElement(se.Name)
          {
             Converted_type = Thrift.ConvertedType.MAP,
-            Num_children = 1
+            Num_children = 1,
+            Repetition_type = Thrift.FieldRepetitionType.OPTIONAL,
          };
          container.Add(root);
 

src/Parquet/File/Values/RunLengthBitPackingHybridValuesWriter.cs

@@ -58,7 +58,7 @@ private static void WriteData(BinaryWriter writer, List<int> data, int bitWidth)
       {
          //for simplicity, we're only going to write RLE, however bitpacking needs to be implemented as well
 
-         const int maxCount = 0b0111_1111_1111_1111;  //max count for an integer with one lost bit
+         const int maxCount = int.MaxValue >> 1;  //max count for an integer with one lost bit
 
          //chunk identical values and write
          int lastValue = 0;

src/spark-experiments/src/main/scala/maps.sc

@@ -25,4 +25,8 @@ val df = sc.parallelize(Seq(
 df.printSchema
 df.show
 
-write(df, "c:\\tmp\\sparkmap.parquet")
+//write(df, "c:\\tmp\\sparkmap.parquet")
+
+val cp = spark.read.parquet("c:\\tmp\\map.parquet")
+cp.printSchema
+cp.show