Comments (16)
Trying to reply within a day, will try to go even faster.
I believe the solution is to make sure your class overrides Equals.
from trill.
Ah! Looking at the code, it appears there may be a bug there. One moment, verifying. I think it's looking at and testing the wrong type.
from trill.
After running the above code, what I surmise is that simply implementing IEquatable<T>
is not sufficient for the check as I had surmised. We actually need the implementation of IEqualityComparerExpression<T>
. Which is, of course, completely non-obvious from the error message or even a cursory look at the code.
The issue, unless @mike-barnett has more info than I, is that to do efficient code generation for columnar mode, we need to be able to inspect the expressions for equality and hash code to be able to manipulate them and inline them into the generated code. Without that ability, we would need to be re-packaging each payload one at a time to do the equality check, which is kind of pointless in columnar.
In debugging this issue, I did find two issues that @mike-barnett and I are going to work on, but I did get code running that works:
`
public class IONRecord_Class_WithEquality : IEqualityComparerExpression<IONRecord_Class_WithEquality>
{
public string Id;
public string Data;
public bool Equals(IONRecord_Class_WithEquality other)
{
if (other == null) return false;
return Id == other.Id && Data == other.Data;
}
public override bool Equals(object obj)
{
return Equals(obj as IONRecord_Class_WithEquality);
}
public override int GetHashCode()
{
unchecked
{
var hashcode = 83;
hashcode = (hashcode * 89) + (this.Id != null ? this.Id.GetHashCode() : 0);
hashcode = (hashcode * 89) + (this.Data != null ? this.Data.GetHashCode() : 0);
return hashcode;
}
}
public static bool operator ==(IONRecord_Class_WithEquality first, IONRecord_Class_WithEquality second) => first.Equals(second);
public static bool operator !=(IONRecord_Class_WithEquality first, IONRecord_Class_WithEquality second) => !first.Equals(second);
public override string ToString() => $" {{{nameof(Id)}:{Id}, {nameof(Data)}:{Data}}}";
public Expression<Func<IONRecord_Class_WithEquality, IONRecord_Class_WithEquality, bool>> GetEqualsExpr()
=> (left, right) => left.Data == right.Data && left.Id == right.Id;
public Expression<Func<IONRecord_Class_WithEquality, int>> GetGetHashCodeExpr()
=> (that) => ((83
* 89) + (that.Id != null ? that.Id.GetHashCode() : 0)
* 89) + (that.Data != null ? that.Data.GetHashCode() : 0);
}
[TestMethod, TestCategory("Gated")]
public void LeftOuterJoin_WithEquality_Test()
{
var assets_subject = new Subject<StreamEvent<IONRecord_Class_WithEquality>>();
var iconnect_subject = new Subject<StreamEvent<IONRecord_Class_WithEquality>>();
var assets_stream = assets_subject.ToStreamable(disorderPolicy: DisorderPolicy.Throw(), flushPolicy: FlushPolicy.FlushOnPunctuation, periodicPunctuationPolicy: null);
var iconnect_stream = iconnect_subject.ToStreamable(disorderPolicy: DisorderPolicy.Throw(), flushPolicy: FlushPolicy.FlushOnPunctuation, periodicPunctuationPolicy: null);
// It shouldn't throw... but is still does
// Type of left side of join, 'xxx', does not have a valid equality operator for columnar mode
var output_stream = assets_stream
.LeftOuterJoin(
right: iconnect_stream,
leftKeySelector: a => a.Id,
rightKeySelector: i => i.Id,
outerResultSelector: a => new
{
a_Id = a.Id,
i_Id = (string)null
},
innerResultSelector: (a, id) => new
{
a_Id = a.Id,
i_Id = id.Id
});
}
`
from trill.
Well, the exception message has been fixed as has the issue you were seeing just above. New NuGet pushed.
from trill.
More details (wanted to get the previous answer quickly out there in case you're online and in a rush):
The Join operator in columnar execution needs to be able to reason about whether payloads are equal, and doing so by reference equality in columnar mode just doesn't work. There are a few ways that Trill recognizes if there is a valid equality "operator":
- If the type implements
IEquatable<T>
(simplest for the implementer) - If the type implements
Microsoft.StreamProcessing.IEqualityComparerExpression<T>
(easiest for Trill to work with for code generation, but by far the hardest for the implementer) - If the type has overridden Equals and GetHashCode methods (least preferable but we'll work with it)
An alternative solution is to turn off columnar execution. You can do that by setting Config.ForceRowBasedExecution to true.
from trill.
from trill.
We (@cybertyche and I) just looked at the code together and we are very unclear on whether there is actually a bug in the code or not. Would it be possible for you to post an (extremely short) repro here so we can dig into it? Sorry for the problem!
from trill.
Try this:
public class IONRecord_Class_WithEquality : IEquatable<IONRecord_Class_WithEquality>
{
public string Id;
public string Data;
public bool Equals(IONRecord_Class_WithEquality other)
{
if (other == null) return false;
return Id == other.Id && Data == other.Data;
}
public override bool Equals(object obj)
{
return Equals(obj as IONRecord_Class_WithEquality);
}
public override int GetHashCode()
{
unchecked
{
var hashcode = 83;
hashcode = (hashcode * 89) + (this.Id != null ? this.Id.GetHashCode() : 0);
hashcode = (hashcode * 89) + (this.Data != null ? this.Data.GetHashCode() : 0);
return hashcode;
}
}
public static bool operator == (IONRecord_Class_WithEquality first, IONRecord_Class_WithEquality second) => first.Equals(second);
public static bool operator != (IONRecord_Class_WithEquality first, IONRecord_Class_WithEquality second) => !first.Equals(second);
public override string ToString() => $" {{{nameof(Id)}:{Id}, {nameof(Data)}:{Data}}}";
}
[Fact]
public void LeftOuterJoin_WithEquality_Test()
{
var assets_subject = new Subject<StreamEvent<IONRecord_Class_WithEquality>>();
var iconnect_subject = new Subject<StreamEvent<IONRecord_Class_WithEquality>>();
var assets_stream = assets_subject.ToStreamable(disorderPolicy: DisorderPolicy.Throw(), flushPolicy: FlushPolicy.FlushOnPunctuation, periodicPunctuationPolicy: null);
var iconnect_stream = iconnect_subject.ToStreamable(disorderPolicy: DisorderPolicy.Throw(), flushPolicy: FlushPolicy.FlushOnPunctuation, periodicPunctuationPolicy: null);
//It shouldn't throw... but is still does
//Type of left side of join, 'xxx', does not have a valid equality operator for columnar mode
Assert.Throws<InvalidOperationException>(() =>
{
var output_stream = assets_stream
.LeftOuterJoin(right: iconnect_stream,
leftKeySelector: a => a.Id,
rightKeySelector: i => i.Id,
outerResultSelector: a => new
{
a_Id = a.Id,
i_Id = (string)null
},
innerResultSelector: (a, i) => new
{
a_Id = a.Id,
i_Id = i.Id
});
});
}
from trill.
BTW, Your fix makes Trill throw a new exception:
using System;
using System.Collections.Generic;
using System.Reactive.Subjects;
using Microsoft.StreamProcessing;
using Microsoft.VisualStudio.TestTools.UnitTesting;
namespace SimpleTesting
{
public class IONRecord_Class : IEquatable<IONRecord_Class>
{
public string Id;
public string Data;
public override bool Equals(object obj)
{
return Equals(obj as IONRecord_Class);
}
public bool Equals(IONRecord_Class other)
{
return other != null &&
Id == other.Id &&
Data == other.Data;
}
public override int GetHashCode()
{
var hashCode = -1123738939;
hashCode = hashCode * -1521134295 + EqualityComparer<string>.Default.GetHashCode(Id);
hashCode = hashCode * -1521134295 + EqualityComparer<string>.Default.GetHashCode(Data);
return hashCode;
}
public override string ToString() => $" {{{nameof(Id)}:{Id}, {nameof(Data)}:{Data}}}";
public static bool operator ==(IONRecord_Class left, IONRecord_Class right)
{
return EqualityComparer<IONRecord_Class>.Default.Equals(left, right);
}
public static bool operator !=(IONRecord_Class left, IONRecord_Class right)
{
return !(left == right);
}
}
public struct IONRecord_Struct
{
public string Id;
public string Data;
public override string ToString() => $" {{{nameof(Id)}:{Id}, {nameof(Data)}:{Data}}}";
}
[TestClass]
public class MyTests
{
[TestMethod]
public void ThrowNullExceptionWithClassPayload_Test()
{
var left_subject = new Subject<StreamEvent<IONRecord_Class>>();
var right_subject = new Subject<StreamEvent<IONRecord_Class>>();
var left_stream = left_subject.ToStreamable(disorderPolicy: DisorderPolicy.Throw(), flushPolicy: FlushPolicy.FlushOnPunctuation, periodicPunctuationPolicy: null);
var right_stream = right_subject.ToStreamable(disorderPolicy: DisorderPolicy.Throw(), flushPolicy: FlushPolicy.FlushOnPunctuation, periodicPunctuationPolicy: null);
var output_events = new List<string>();
var output_stream = left_stream
.LeftOuterJoin(
right: right_stream,
leftKeySelector: l => l.Id,
rightKeySelector: r => r.Id,
outerResultSelector: l => new
{
l_Id = l.Id,
r_Id = (string)null
},
innerResultSelector: (l, r) => new
{
l_Id = l.Id,
r_Id = r.Id
});
var output_observable = output_stream.ToStreamEventObservable(reshapingPolicy: ReshapingPolicy.None);
output_observable.Subscribe(onNext: se =>
{
if (se.IsData) output_events.Add(se.ToString());
});
left_subject.OnNext(StreamEvent.CreateStart(701, new IONRecord_Class { Id = "c1", Data = "shortdes1" }));
// NEXT LINE THROWS!!
left_subject.OnNext(StreamEvent.CreatePunctuation<IONRecord_Class>(punctuationTime: 702));
}
[TestMethod]
public void DoesNOTThrowWithStructPayload_Test()
{
var left_subject = new Subject<StreamEvent<IONRecord_Struct>>();
var right_subject = new Subject<StreamEvent<IONRecord_Struct>>();
var left_stream = left_subject.ToStreamable(disorderPolicy: DisorderPolicy.Throw(), flushPolicy: FlushPolicy.FlushOnPunctuation, periodicPunctuationPolicy: null);
var right_stream = right_subject.ToStreamable(disorderPolicy: DisorderPolicy.Throw(), flushPolicy: FlushPolicy.FlushOnPunctuation, periodicPunctuationPolicy: null);
var output_events = new List<string>();
var output_stream = left_stream
.LeftOuterJoin(
right: right_stream,
leftKeySelector: l => l.Id,
rightKeySelector: r => r.Id,
outerResultSelector: l => new
{
l_Id = l.Id,
r_Id = (string)null
},
innerResultSelector: (l, r) => new
{
l_Id = l.Id,
r_Id = r.Id
});
var output_observable = output_stream.ToStreamEventObservable(reshapingPolicy: ReshapingPolicy.None);
output_observable.Subscribe(onNext: se =>
{
if (se.IsData) output_events.Add(se.ToString());
});
left_subject.OnNext(StreamEvent.CreateStart(701, new IONRecord_Struct { Id = "c1", Data = "shortdes1" }));
left_subject.OnNext(StreamEvent.CreatePunctuation<IONRecord_Struct>(punctuationTime: 702));
right_subject.OnNext(StreamEvent.CreatePunctuation<IONRecord_Struct>(punctuationTime: 702));
// yay.. above code didnt throw
Assert.IsTrue(true);
}
}
}
from trill.
PR for fix incoming.
from trill.
Fix pushed to NuGet.
from trill.
from trill.
I'm not seeing the cast exception when running your code locally. Instead, I'm seeing it throw due to the class implementing IEquatable instead of IEqualityComparerExpression. The invalid cast you're seeing is consistent with behavior before the NuGet published a few hours ago.
from trill.
Never mind, I can repro - I had test settings set up differently than you. One moment.
from trill.
from trill.
from trill.
Related Issues (20)
- Install trill on a Linux server?
- Output when SessionTimeoutWindow() ends HOT 11
- NullReferenceException in Columnar Mode HOT 4
- [QUESTION] Output count = 0 when the last event leaves snapshot window HOT 2
- Controlled flush on partitioned streams HOT 1
- Output on aggregations and partition reset
- ReflectionSchemaBuilder ignores knownTypes of the nested type
- Parsing query language to derive Trill operation HOT 1
- NullReferenceException when using checkpointing HOT 2
- Random splitter in a multicore setting
- Trill with Azure Storage Blob and Table HOT 1
- Ingress from Azure Event Hubs - how?
- NullReferenceException when running under net6.0 HOT 1
- Is Trill alive? HOT 10
- Enable .NET 7.0 support HOT 6
- Trillsx HOT 1
- Calling Connect() on a ConnectableStreamable without subscribers throw System.NullReferenceException
- This repo is missing important files HOT 1
- .NET 8 support
- NullReferenceException in .NET 8
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from trill.