Skip to content

Commit 17c2a7f

Browse files
author
Ivan Gavryliuk
authored
support dates (INT64 only) in plain encoding (tensorflow#82)
* support dates (INT64 only) in plain encoding * update readme
1 parent c40c502 commit 17c2a7f

10 files changed

Lines changed: 202 additions & 95 deletions

File tree

‎README.md‎

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -39,17 +39,18 @@ For instance, to read a file `c:\test.parquet` you woudl normally write the foll
3939
```csharp
4040
using System.IO;
4141
using Parquet;
42+
using Parquet.Data;
4243

4344
using(Stream fs = File.OpenRead("c:\\test.parquet"))
4445
{
4546
using(var reader = new ParquetReader(fs))
4647
{
47-
ParquetDataSet ds = reader.Read();
48+
DataSet ds = reader.Read();
4849
}
4950
}
5051
```
5152

52-
this will read entire file in memory as a set of columns inside `ParquetDataSet` class.
53+
this will read entire file in memory as a set of rows inside `DataSet` class.
5354

5455
### Writing files
5556

@@ -58,20 +59,21 @@ Parquet.Net operates on streams, therefore you need to create it first. The foll
5859
```csharp
5960
using System.IO;
6061
using Parquet;
62+
using Parquet.Data;
6163

62-
var idColumn = new ParquetColumn<int>("id");
63-
idColumn.Add(1, 2);
64+
var ds = new DataSet(
65+
new SchemaElement<int>("id", false),
66+
new SchemaElement<string>("city", false)
67+
);
6468

65-
var cityColumn = new ParquetColumn<string>("city");
66-
cityColumn.Add("London", "Derby");
67-
68-
var dataSet = new ParquetDataSet(idColumn, cityColumn);
69+
ds.Add(1, "London");
70+
ds.Add(2, "Derby");
6971

7072
using(Stream fileStream = File.OpenWrite("c:\\test.parquet"))
7173
{
7274
using(var writer = new ParquetWriter(fileStream))
7375
{
74-
writer.Write(dataSet);
76+
writer.Write(ds);
7577
}
7678
}
7779

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

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
using Parquet.Data;
22
using Parquet.File.Values;
3+
using System;
34
using System.Collections.Generic;
45
using System.IO;
56
using System.Linq;
@@ -29,7 +30,7 @@ public PlainValuesReaderWriterTest()
2930
}
3031

3132
[Fact]
32-
public void Booleans()
33+
public void Array_of_booleans_writes_and_reads()
3334
{
3435
var bools = new List<bool>();
3536
bools.AddRange(new[] { true, false, true });
@@ -45,5 +46,21 @@ public void Booleans()
4546

4647
Assert.Equal(bools, boolsRead.Cast<bool>().ToList());
4748
}
49+
50+
[Fact]
51+
public void DateTimeOffset_writes_and_reads()
52+
{
53+
var dates = new List<DateTimeOffset> { DateTime.UtcNow.RoundToSecond() }; //this version doesn't store milliseconds
54+
var schema = new SchemaElement<DateTimeOffset>("dto", false);
55+
56+
_writer.Write(_bw, schema, dates);
57+
58+
_ms.Position = 0;
59+
60+
var datesRead = new List<DateTimeOffset>();
61+
_reader.Read(_br, schema, datesRead, 1);
62+
63+
Assert.Equal(datesRead[0], dates[0]);
64+
}
4865
}
4966
}

‎src/Parquet.Test/Reader/ParquetCsvComparison.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,7 @@ private object ChangeType(object v, Type t)
104104
private DataSet ReadParquet(string name)
105105
{
106106
string path = GetDataFilePath(name);
107-
return ParquetReader.ReadFile(path);
107+
return ParquetReader.ReadFile(path, new ParquetOptions { TreatByteArrayAsString = true });
108108
}
109109

110110
private DataSet ReadCsv(string name)

‎src/Parquet/Data/SchemaElement.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,12 @@ internal SchemaElement(PSE thriftSchema)
7272

7373
internal PSE Thrift { get; set; }
7474

75+
internal bool IsAnnotatedWith(Thrift.ConvertedType ct)
76+
{
77+
//checking __isset is important, this is a way of Thrift to tell whether the variable is set at all
78+
return Thrift.__isset.converted_type && Thrift.Converted_type == ct;
79+
}
80+
7581
/// <summary>
7682
/// Pretty prints
7783
/// </summary>

src/Parquet/File/ListFactory.cs renamed to src/Parquet/File/TypeFactory.cs

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
using System.Collections;
44
using System.Reflection;
55
using Parquet.Data;
6+
using System.Linq;
67

78
namespace Parquet.File
89
{
@@ -25,13 +26,18 @@ public TypeTag(Thrift.Type ptype, Thrift.ConvertedType? convertedType)
2526
{
2627
{ typeof(int), new TypeTag(Thrift.Type.INT32, null) },
2728
{ typeof(bool), new TypeTag(Thrift.Type.BOOLEAN, null) },
28-
{ typeof(string), new TypeTag(Thrift.Type.BYTE_ARRAY, Thrift.ConvertedType.UTF8) }
29+
{ typeof(string), new TypeTag(Thrift.Type.BYTE_ARRAY, Thrift.ConvertedType.UTF8) },
30+
{ typeof(DateTimeOffset), new TypeTag(Thrift.Type.INT64, Thrift.ConvertedType.TIMESTAMP_MILLIS) }
2931
};
3032

3133
public static void AdjustSchema(Thrift.SchemaElement schema, Type systemType)
3234
{
3335
if (!TypeToTag.TryGetValue(systemType, out TypeTag tag))
34-
throw new NotSupportedException($"system type {systemType} is not supported");
36+
{
37+
string supportedTypes = string.Join(", ", TypeToTag.Keys.Select(t => t.ToString()));
38+
39+
throw new NotSupportedException($"system type {systemType} is not supported, we currently support only these types: '{supportedTypes}'");
40+
}
3541

3642
schema.Type = tag.PType;
3743

@@ -64,6 +70,7 @@ public static IList Create(SchemaElement schema, bool nullable = false)
6470
return Create(t, nullable);
6571
}
6672

73+
//todo: this can be rewritten by looking up in TypeToTag
6774
public static Type ToSystemType(Thrift.SchemaElement schema)
6875
{
6976
switch (schema.Type)

‎src/Parquet/File/Values/PlainValuesReader.cs‎

Lines changed: 22 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
using System.Runtime.CompilerServices;
99
using System.Numerics;
1010
using Parquet.Data;
11+
using System.Collections.Generic;
1112

1213
namespace Parquet.File.Values
1314
{
@@ -81,7 +82,7 @@ private static void ReadPlainBoolean(byte[] data, IList destination, long maxVal
8182
[MethodImpl(MethodImplOptions.AggressiveInlining)]
8283
private static void ReadInt32(byte[] data, SchemaElement schema, IList destination)
8384
{
84-
if (schema.Thrift.Converted_type == Thrift.ConvertedType.DATE)
85+
if (schema.IsAnnotatedWith(Thrift.ConvertedType.DATE))
8586
{
8687
for (int i = 0; i < data.Length; i += 4)
8788
{
@@ -112,10 +113,23 @@ private static void ReadFloat(byte[] data, SchemaElement schema, IList destinati
112113
[MethodImpl(MethodImplOptions.AggressiveInlining)]
113114
private static void ReadLong(byte[] data, SchemaElement schema, IList destination)
114115
{
115-
for (int i = 0; i < data.Length; i += 8)
116+
if (schema.IsAnnotatedWith(Thrift.ConvertedType.TIMESTAMP_MILLIS))
116117
{
117-
long lv = BitConverter.ToInt64(data, i);
118-
destination.Add(lv);
118+
var lst = (List<DateTimeOffset>)destination;
119+
120+
for (int i = 0; i < data.Length; i += 8)
121+
{
122+
long lv = BitConverter.ToInt64(data, i);
123+
lst.Add(lv.FromUnixTime());
124+
}
125+
}
126+
else
127+
{
128+
for (int i = 0; i < data.Length; i += 8)
129+
{
130+
long lv = BitConverter.ToInt64(data, i);
131+
destination.Add(lv);
132+
}
119133
}
120134
}
121135

@@ -134,7 +148,7 @@ private static void ReadFixedLenByteArray(byte[] data, SchemaElement schema, ILi
134148
{
135149
for (int i = 0; i < data.Length; i += schema.Thrift.Type_length)
136150
{
137-
if (schema.Thrift.Converted_type != Thrift.ConvertedType.DECIMAL) continue;
151+
if (!schema.IsAnnotatedWith(Thrift.ConvertedType.DECIMAL)) continue;
138152
// go from data - decimal needs to be 16 bytes but not from Spark - variable fixed nonsense
139153
byte[] dataNew = new byte[schema.Thrift.Type_length];
140154
Array.Copy(data, i, dataNew, 0, schema.Thrift.Type_length);
@@ -169,7 +183,7 @@ private static void ReadInt96(byte[] data, SchemaElement schema, IList destinati
169183
byte[] nanos = new byte[8];
170184
Array.Copy(data, i + 8, v96, 0, 4);
171185
Array.Copy(data, i, nanos, 0, 8);
172-
var bi = BitConverter.ToInt32(v96, 0).JulianToDateTime();
186+
DateTime bi = BitConverter.ToInt32(v96, 0).JulianToDateTime();
173187
long nanosToInt64 = BitConverter.ToInt64(nanos, 0);
174188
double millis = (double) nanosToInt64 / 1000000D;
175189
bi = bi.AddMilliseconds(millis);
@@ -185,8 +199,8 @@ private void ReadByteArray(byte[] data, SchemaElement schemaElement, IList desti
185199
// Both UTF8 and JSON are stored as binary data (byte_array) which allows annotations to be used either UTF8 and JSON
186200
// They should be treated in the same way as Strings
187201
// need to find a better implementation for this but date strings are always broken here because of the type mismatch
188-
if (schemaElement.Thrift.__isset.converted_type ||
189-
schemaElement.Thrift.Converted_type == Thrift.ConvertedType.UTF8 || schemaElement.Thrift.Converted_type == Thrift.ConvertedType.JSON ||
202+
if (schemaElement.IsAnnotatedWith(Thrift.ConvertedType.UTF8) ||
203+
schemaElement.IsAnnotatedWith(Thrift.ConvertedType.JSON) ||
190204
_options.TreatByteArrayAsString)
191205
{
192206
for (int i = 0; i < data.Length;)

‎src/Parquet/File/Values/PlainValuesWriter.cs‎

Lines changed: 40 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -44,8 +44,9 @@ public void Write(BinaryWriter writer, SchemaElement schema, IList data)
4444
WriteDouble(writer, schema, data);
4545
break;
4646

47-
//case TType.INT96:
48-
// break;
47+
case TType.INT96:
48+
WriteInt96(writer, schema, data);
49+
break;
4950

5051
case TType.BYTE_ARRAY:
5152
WriteByteArray(writer, schema, data);
@@ -91,10 +92,22 @@ private static void WriteBoolean(BinaryWriter writer, SchemaElement schema, ILis
9192
[MethodImpl(MethodImplOptions.AggressiveInlining)]
9293
private static void WriteInt32(BinaryWriter writer, SchemaElement schema, IList data)
9394
{
94-
var dataTyped = (List<int>)data;
95-
foreach (int el in dataTyped)
95+
if (schema.IsAnnotatedWith(Thrift.ConvertedType.DATE))
96+
{
97+
var dataTyped = (List<DateTimeOffset>)data;
98+
foreach(DateTimeOffset el in dataTyped)
99+
{
100+
int days = (int)el.ToUnixDays();
101+
writer.Write(days);
102+
}
103+
}
104+
else
96105
{
97-
writer.Write(el);
106+
var dataTyped = (List<int>)data;
107+
foreach (int el in dataTyped)
108+
{
109+
writer.Write(el);
110+
}
98111
}
99112
}
100113

@@ -111,13 +124,31 @@ private static void WriteFloat(BinaryWriter writer, SchemaElement schema, IList
111124
[MethodImpl(MethodImplOptions.AggressiveInlining)]
112125
private static void WriteLong(BinaryWriter writer, SchemaElement schema, IList data)
113126
{
114-
var lst = (List<long>)data;
115-
foreach (long l in lst)
127+
if (schema.IsAnnotatedWith(Thrift.ConvertedType.TIMESTAMP_MILLIS))
128+
{
129+
var lst = (List<DateTimeOffset>)data;
130+
foreach(DateTimeOffset dto in lst)
131+
{
132+
long unixTime = dto.ToUnixTime();
133+
writer.Write(unixTime);
134+
}
135+
}
136+
else
116137
{
117-
writer.Write(l);
138+
var lst = (List<long>)data;
139+
foreach (long l in lst)
140+
{
141+
writer.Write(l);
142+
}
118143
}
119144
}
120145

146+
[MethodImpl(MethodImplOptions.AggressiveInlining)]
147+
private void WriteInt96(BinaryWriter writer, SchemaElement schema, IList data)
148+
{
149+
throw new NotImplementedException();
150+
}
151+
121152
[MethodImpl(MethodImplOptions.AggressiveInlining)]
122153
private static void WriteDouble(BinaryWriter writer, SchemaElement schema, IList data)
123154
{
@@ -127,6 +158,7 @@ private static void WriteDouble(BinaryWriter writer, SchemaElement schema, IList
127158
writer.Write(d);
128159
}
129160
}
161+
130162
[MethodImpl(MethodImplOptions.AggressiveInlining)]
131163
private void WriteByteArray(BinaryWriter writer, SchemaElement schema, IList data)
132164
{

‎src/Parquet/ParquetOptions.cs‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,24 @@
55
/// </summary>
66
public class ParquetOptions
77
{
8+
/// <summary>
9+
/// Initializes a new instance of the <see cref="ParquetOptions"/> class.
10+
/// </summary>
11+
public ParquetOptions()
12+
{
13+
TreatByteArrayAsString = false;
14+
15+
TreatBigIntegersAsDates = true;
16+
}
17+
818
/// <summary>
919
/// When true byte arrays will be treated as UTF-8 strings
1020
/// </summary>
1121
public bool TreatByteArrayAsString { get; set; }
22+
23+
/// <summary>
24+
/// Gets or sets a value indicating whether big integers are always treated as dates
25+
/// </summary>
26+
public bool TreatBigIntegersAsDates { get; set; }
1227
}
1328
}

‎src/Parquet/ParquetReader.cs‎

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -39,13 +39,14 @@ public class ParquetReader : IDisposable
3939
private readonly BinaryReader _reader;
4040
private readonly ThriftStream _thrift;
4141
private Thrift.FileMetaData _meta;
42-
private readonly ParquetOptions _options = new ParquetOptions();
42+
private readonly ParquetOptions _options;
4343

4444
/// <summary>
4545
/// Creates an instance from input stream
4646
/// </summary>
47-
/// <param name="input"></param>
48-
public ParquetReader(Stream input)
47+
/// <param name="input">Input stream, must be readable and seekable</param>
48+
/// <param name="options">Optional reader options</param>
49+
public ParquetReader(Stream input, ParquetOptions options = null)
4950
{
5051
_input = input ?? throw new ArgumentNullException(nameof(input));
5152
if (!input.CanRead || !input.CanSeek) throw new ArgumentException("stream must be readable and seekable", nameof(input));
@@ -54,13 +55,20 @@ public ParquetReader(Stream input)
5455
_reader = new BinaryReader(input);
5556
ValidateFile();
5657
_thrift = new ThriftStream(input);
58+
_options = options ?? new ParquetOptions();
5759
}
5860

59-
public static DataSet ReadFile(string fullPath)
61+
/// <summary>
62+
/// Reads the file
63+
/// </summary>
64+
/// <param name="fullPath">The full path.</param>
65+
/// <param name="options">Optional reader options.</param>
66+
/// <returns>DataSet</returns>
67+
public static DataSet ReadFile(string fullPath, ParquetOptions options = null)
6068
{
6169
using (Stream fs = System.IO.File.OpenRead(fullPath))
6270
{
63-
using (var reader = new ParquetReader(fs))
71+
using (var reader = new ParquetReader(fs, options))
6472
{
6573
return reader.Read();
6674
}

0 commit comments

Comments
 (0)