Skip to content

Update Apache.Arrow reference to v0.14.#167

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
imback82 merged 5 commits into from
Jul 23, 2019
Merged

Update Apache.Arrow reference to v0.14. #167

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6cbb710
Update Apache.Arrow reference to v0.14.
eerhardt Jul 9, 2019
6835e7f
Merge branch 'master' into UpdateArrow
imback82 Jul 10, 2019
adbfec5
Fix Tpch build.
eerhardt Jul 10, 2019
b89c030
Merge remote-tracking branch 'upstream/master' into UpdateArrow
eerhardt Jul 23, 2019
79cdefd
Update to Apache.Arrow 0.14.1.
eerhardt Jul 23, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions benchmark/csharp/Tpch/TpchFunctionalQueries.cs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -10,6 +10,7 @@
using Microsoft.Spark.Sql;
using static Microsoft.Spark.Sql.ExperimentalFunctions;
using static Microsoft.Spark.Sql.Functions;
using Column = Microsoft.Spark.Sql.Column;

namespace Tpch
{
Expand Down
18 changes: 4 additions & 14 deletions benchmark/csharp/Tpch/VectorFunctions.cs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,7 +17,7 @@ internal static DoubleArray ComputeTotal(DoubleArray price, DoubleArray discount
}

int length = price.Length;
var builder = new ArrowBuffer.Builder<double>(length);
var builder = new DoubleArray.Builder().Reserve(length);
ReadOnlySpan<double> prices = price.Values;
ReadOnlySpan<double> discounts = discount.Values;
ReadOnlySpan<double> taxes = tax.Values;
Expand All@@ -26,12 +26,7 @@ internal static DoubleArray ComputeTotal(DoubleArray price, DoubleArray discount
builder.Append(prices[i] * (1 - discounts[i]) * (1 + taxes[i]));
}

return new DoubleArray(
builder.Build(),
nullBitmapBuffer: ArrowBuffer.Empty,
length: length,
nullCount: 0,
offset: 0);
return builder.Build();
}

internal static DoubleArray ComputeDiscountPrice(DoubleArray price, DoubleArray discount)
Expand All@@ -42,20 +37,15 @@ internal static DoubleArray ComputeDiscountPrice(DoubleArray price, DoubleArray
}

int length = price.Length;
var builder = new ArrowBuffer.Builder<double>(length);
var builder = new DoubleArray.Builder().Reserve(length);
ReadOnlySpan<double> prices = price.Values;
ReadOnlySpan<double> discounts = discount.Values;
for (int i = 0; i < length; ++i)
{
builder.Append(prices[i] * (1 - discounts[i]));
}

return new DoubleArray(
builder.Build(),
nullBitmapBuffer: ArrowBuffer.Empty,
length: length,
nullCount: 0,
offset: 0);
return builder.Build();
}
}
}
9 changes: 2 additions & 7 deletions examples/Microsoft.Spark.CSharp.Examples/Sql/VectorUdfs.cs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -80,18 +80,13 @@ private static RecordBatch CountCharacters(

return new RecordBatch(
new Schema.Builder()
.Field(f => f.Name(groupField.Name).DataType(groupField.DataType))
.Field(groupField)
.Field(f => f.Name(stringFieldName + "_CharCount").DataType(Int32Type.Default))
.Build(),
new IArrowArray[]
{
records.Column(groupFieldIndex),
new Int32Array(
new ArrowBuffer.Builder<int>().Append(characterCount).Build(),
ArrowBuffer.Empty,
length: 1,
nullCount: 0,
offset: 0)
new Int32Array.Builder().Append(characterCount).Build()
},
returnLength);
}
Expand Down
22 changes: 7 additions & 15 deletions src/csharp/Microsoft.Spark.E2ETest/IpcTests/Sql/DataFrameTests.cs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -11,6 +11,7 @@
using Xunit;
using static Microsoft.Spark.Sql.Functions;
using static Microsoft.Spark.UnitTest.TestUtils.ArrowTestUtils;
using Column = Microsoft.Spark.Sql.Column;
using Int32Type = Apache.Arrow.Types.Int32Type;

namespace Microsoft.Spark.E2ETest.IpcTests
Expand DownExpand Up@@ -190,16 +191,12 @@ public void TestGroupedMapUdf()
Assert.Equal(3, rows.Length);
foreach (Row row in rows)
{
int age = row.GetAs<int>("age");
int? charCount = row.GetAs<int?>("nameCharCount");
int? age = row.GetAs<int?>("age");
int charCount = row.GetAs<int>("nameCharCount");
switch (age)
{
case 0:
// The results here are incorrect for the {name: "Michael" age: null}
// record because of https://issues.apache.org/jira/browse/ARROW-5887.
// When an updated Apache.Arrow library is available with the fix,
// this should change to check for age: null, charCount: 7.
Assert.Null(charCount);
case null:
Assert.Equal(7, charCount);
break;
case 19:
Assert.Equal(11, charCount);
Expand DownExpand Up@@ -234,18 +231,13 @@ private static RecordBatch CountCharacters(RecordBatch records)

return new RecordBatch(
new Schema.Builder()
.Field(f => f.Name(groupField.Name).DataType(groupField.DataType))
.Field(groupField)
.Field(f => f.Name("name_CharCount").DataType(Int32Type.Default))
.Build(),
new IArrowArray[]
{
records.Column(groupFieldIndex),
new Int32Array(
new ArrowBuffer.Builder<int>().Append(characterCount).Build(),
ArrowBuffer.Empty,
length: 1,
nullCount: 0,
offset: 0)
new Int32Array.Builder().Append(characterCount).Build()
},
returnLength);
}
Expand Down
2 changes: 1 addition & 1 deletion src/csharp/Microsoft.Spark.UnitTest/WorkerFunctionTests.cs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -93,7 +93,7 @@ public void TestArrowWorkerFunctionForBool()
new ArrowUdfWrapper<StringArray, BooleanArray, BooleanArray>(
(strings, flags) => (BooleanArray)ToArrowArray(
Enumerable.Range(0, strings.Length)
.Select(i => flags.GetBoolean(i).Value || strings.GetString(i).Contains("true"))
.Select(i => flags.GetBoolean(i) || strings.GetString(i).Contains("true"))
.ToArray())).Execute);

IArrowArray[] input = new[]
Expand Down
32 changes: 2 additions & 30 deletions src/csharp/Microsoft.Spark.Worker/Command/SqlCommandExecutor.cs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -291,9 +291,6 @@ public object Run(int splitId, object input)
/// </summary>
internal class ArrowSqlCommandExecutor : SqlCommandExecutor
{
[ThreadStatic]
private static MemoryStream s_writeOutputStream;

protected override CommandExecutorStat ExecuteCore(
Stream inputStream,
Stream outputStream,
Expand All@@ -304,11 +301,6 @@ protected override CommandExecutorStat ExecuteCore(

SerDe.Write(outputStream, (int)SpecialLengths.START_ARROW_STREAM);

// TODO: Remove this MemoryStream once the arrow writer supports non-seekable streams.
// For now, we write to a temporary seekable MemoryStream which we then copy to
// the actual destination stream.
MemoryStream tmp = s_writeOutputStream ?? (s_writeOutputStream = new MemoryStream());

ArrowStreamWriter writer = null;
Schema resultSchema = null;
foreach (ReadOnlyMemory<IArrowArray> input in GetInputIterator(inputStream))
Expand All@@ -319,24 +311,18 @@ protected override CommandExecutorStat ExecuteCore(
int numEntries = results[0].Length;
stat.NumEntriesProcessed += numEntries;

tmp.SetLength(0);

if (writer == null)
{
Debug.Assert(resultSchema == null);
resultSchema = BuildSchema(results);

writer = new ArrowStreamWriter(tmp, resultSchema, leaveOpen: true);
writer = new ArrowStreamWriter(outputStream, resultSchema, leaveOpen: true);
}

var recordBatch = new RecordBatch(resultSchema, results, numEntries);

// TODO: Remove sync-over-async once WriteRecordBatch exists.
writer.WriteRecordBatchAsync(recordBatch).GetAwaiter().GetResult();

tmp.Position = 0;
tmp.CopyTo(outputStream);
outputStream.Flush();
}

SerDe.Write(outputStream, 0);
Expand DownExpand Up@@ -532,9 +518,6 @@ public IArrowArray[] Run(ReadOnlyMemory<IArrowArray> input)

internal class ArrowGroupedMapCommandExecutor : SqlCommandExecutor
{
[ThreadStatic]
private static MemoryStream s_writeOutputStream;

protected override CommandExecutorStat ExecuteCore(
Stream inputStream,
Stream outputStream,
Expand All@@ -548,11 +531,6 @@ protected override CommandExecutorStat ExecuteCore(

SerDe.Write(outputStream, (int)SpecialLengths.START_ARROW_STREAM);

// TODO: Remove this MemoryStream once the arrow writer supports non-seekable streams.
// For now, we write to a temporary seekable MemoryStream which we then copy to
// the actual destination stream.
MemoryStream tmp = s_writeOutputStream ?? (s_writeOutputStream = new MemoryStream());

ArrowStreamWriter writer = null;
foreach (RecordBatch input in GetInputIterator(inputStream))
{
Expand All@@ -561,19 +539,13 @@ protected override CommandExecutorStat ExecuteCore(
int numEntries = result.Length;
stat.NumEntriesProcessed += numEntries;

tmp.SetLength(0);

if (writer == null)
{
writer = new ArrowStreamWriter(tmp, result.Schema, leaveOpen: true);
writer = new ArrowStreamWriter(outputStream, result.Schema, leaveOpen: true);
}

// TODO: Remove sync-over-async once WriteRecordBatch exists.
writer.WriteRecordBatchAsync(result).GetAwaiter().GetResult();

tmp.Position = 0;
tmp.CopyTo(outputStream);
outputStream.Flush();
}

SerDe.Write(outputStream, 0);
Expand Down
2 changes: 1 addition & 1 deletion src/csharp/Microsoft.Spark/Microsoft.Spark.csproj
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -22,7 +22,7 @@
</ItemGroup>

<ItemGroup>
<PackageReference Include="Apache.Arrow" Version="0.13.0" />
<PackageReference Include="Apache.Arrow" Version="0.14.1" />
<PackageReference Include="Microsoft.CSharp" Version="4.5.0" />
<PackageReference Include="Newtonsoft.Json" Version="11.0.2" />
<PackageReference Include="Razorvine.Pyrolite" Version="4.26.0" />
Expand Down