The Problem
Much like a Merge Join in SSIS, the Kimball Method SCD component matches incoming rows from different inputs to each other (where possible). One of the restrictions of participating in the SSIS data flow pipeline is that the component itself can't control which of the inputs is delivered rows. This means that even when the rows are sorted - like the way the Merge Join requires - the probability exists that you'll get a bunch of rows from one of the inputs and have to hold on to them before you get matching rows from the other input. In essence, the component has to "cache" incoming rows.
Unlike the Merge Join component, the Kimball SCD component doesn't require sorted inputs; you can deliver rows to the Kimball SCD in unsorted order. The benefit of requiring sorted inputs is that you can imply that a row from one input will have no possible match on the other input if rows have sorted column values that occur "later" in the sort. For example, you can guarantee that a row with a join key of "B" will never match a row coming in the other input if you've seen a "C" on that input.
The Kimball SCD component does this kind of optimization as well - but it also has to do the more mundane process of simply cachine everything until it finds a match. But as I said, adding to a generic cache is relatively mundane. It's all the other aspects of caching - retrieval, for example - that require more thought. Specifically, these properties were must-haves for my caching mechanism:
- It must be thread-safe for readers and writers. There will be many threads attempting to add and remove items from a single cache construct.
- Key/value/arrival order triples are to be stored. (Arrival order being the ordinal position of the row from the input stream - i.e.: "this is the fifth row received.")
- It must permit items to be interrogated or removed by key.
- It must permit items to be interrogated or removed by their arrival order.
- It must return subsets by specification of a range of arrival orders. (i.e.: "give me any rows that arrived after the fifth row, but before the tenth.")
- All operations should be O(1), or O(log n) at worst - no O(n). The collections could potentially contain millions of items, and O(n) operations cost dearly at that scale.
Not very many .Net collections could represent those requirements alone - especially #2. Dictionaries and Hashtables have key and value "pair" concepts - but nothing has a "triplet" concept. It gets even more difficult when the last item needs to be taken care of - a great deal of the internal collections have O(n) addition, removal and/or interrogation penalties to pay.
The Plan
This would be a little more complicated than a simple combination of stock collections.
In order to store the triplets, I created an "entry" construct that kept the triplet as an atomic unit. My collection would store those "entries". In order to be able to retrieve any entry quickly by it's "key", I stored those entry constructs in a Hashtable, keyed by the "key." In order to retrieve any entry quickly by its arrival order, I stored the keys in a Hashtable - the "key" was the value, and the arrival order was the key. A double lookup allowed retrieval of a "value" by the arrival order. In order to be able to retrieve a set of entries quickly by a range of arrival orders, I kept a List of keys in the order they arrived. To mitigate the performance hit of removing items from the List, I didn't remove items from that List immediately. Instead, I added them to a Hashtable of "deleted keys" that I could use to modify the results of other operations in an O(1) manner. I then intersected the List and Hashtable of deleted items to rebuild the List - an infrequent O(n) operation - when I needed to retrieve a range of results. I think there's still room for improvement here.
The Result
What I ended up with was a very well performing construct for the requirements. It's very specific to my intended use, and most common operations achieve O(1) performance.
The OrderedHashList:
- Has O(1) behaviour on adds
- Has O(1) behaviour on inquiries by key or arrival order
- Has O(log n) behaviour on removes
- Has O(n) behaviour on some range retrievals, O(log n) on others
- Encapsulates all required thread safety - no external locking via a public SyncRoot-type property is required.
#region CLASS: OrderedHashList<K, V>
/// <summary>
/// The OrderedHashList provides a HashTable-like collection of key-value pairs, but extends
/// that capability to allow an "order" to be associated with eash key-value pair. The resulting
/// key-value-order pairs can be retrieved by key or order, and can be enumerated like a List.
/// </summary>
public class OrderedHashList<K, V>
{
#region CLASS: OrderedHashListEntry
private class OrderedHashListEntry
{
#region Private Variables
private K _key;
private V _value;
private int _arrivalOrder;
#endregion
#region Constructor
public OrderedHashListEntry(K key, V value, int arrivalOrder)
{
this._key = key;
this._value = value;
this._arrivalOrder = arrivalOrder;
}
#endregion
#region Public Properties
public K Key
{
get { return this._key; }
}
public V Value
{
get { return this._value; }
}
public int ArrivalOrder
{
get { return this._arrivalOrder; }
}
#endregion
public override string ToString()
{
return "[" + this._key.ToString() + "].[" + this._arrivalOrder.ToString() + "]"
+ ".[" + this._value.ToString() + "]";
}
}
#endregion
#region Private Variables
private readonly object _lock = new object();
private Hashtable _entriesByKey;
private Hashtable _keysByArrivalOrder;
private List<K> _keysInArrivalOrder;
private Hashtable _deletedKeysInArrivalOrder;
#endregion
#region Constructor
public OrderedHashList()
{
this._entriesByKey = new Hashtable();
this._keysByArrivalOrder = new Hashtable();
this._keysInArrivalOrder = new List<K>();
this._deletedKeysInArrivalOrder = new Hashtable();
}
#endregion
#region Public Properties
/// <summary>Returns the number of OrderedHashListEntries contained
/// in the collection. This is an O(1) operation.</summary>
public int Count
{
get { lock (this._lock) { return this._entriesByKey.Count; } }
}
/// <summary>Returns a complete list of values contained in the
/// collection. This is an O(n) operation.</summary>
public List<V> Values
{
get
{
List<V> values = new List<V>();
lock (this._lock)
{
foreach (OrderedHashListEntry entry in this._entriesByKey.Values)
{
values.Add(entry.Value);
}
}
return values;
}
}
#endregion
#region Public Methods
#region Clear
/// <summary>Empties the OrderedHashList. This is an O(1) operation.</summary>
public void Clear()
{
lock (this._lock)
{
this._entriesByKey.Clear();
this._keysByArrivalOrder.Clear();
this._keysInArrivalOrder.Clear();
this._deletedKeysInArrivalOrder.Clear();
}
}
#endregion
#region Add
/// <summary>Adds a key-value-order to the OrderedHashList. An exception
/// will be thrown if the key or arrivalOrder is already present in the collection.
/// This is an O(1) operation.</summary>
/// <param name="key"></param>
/// <param name="value"></param>
//// <param name="arrivalOrder"></param>
public void Add(K key, V value, int arrivalOrder)
{
OrderedHashListEntry entry = new OrderedHashList<K, V>.OrderedHashListEntry(key, value, arrivalOrder);
lock (this._lock)
{
if (this._entriesByKey.Contains(key))
{
#region Report key already exists
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new InvalidOperationException("Key (" + key.ToString() + ") already exists in the collection");
#endregion
}
if (this._keysByArrivalOrder.ContainsKey(arrivalOrder))
{
#region Report order already exists
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new InvalidOperationException("Arrival order (" + arrivalOrder.ToString()
+ ") already exists in the collection");
#endregion
}
try
{
this._entriesByKey.Add(key, entry);
this._keysByArrivalOrder.Add(entry.ArrivalOrder, entry.Key);
this._keysInArrivalOrder.Add(entry.Key);
}
#region catch ...
catch (Exception ex)
{
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Unexpected exception in OrderedHashList.Add: "
+ ex.Message, ex);
}
#endregion
}
}
#endregion
#region Remove
/// <summary>Removes the entry with the specified key from the collection.
/// This is an O(log n) operation.</summary>
/// <param name="key"></param>
public void Remove(K key)
{
lock (this._lock)
{
if (this._entriesByKey.ContainsKey(key))
{
OrderedHashListEntry entry;
entry = (OrderedHashListEntry)this._entriesByKey[key];
if (entry != null)
{
try
{
if (this._entriesByKey.ContainsKey(entry.Key))
{
this._entriesByKey.Remove(entry.Key);
}
else
{
#region Report internal corruption
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Internal collection corruption detected in "
+ "OrderedHashList.Remove. Key (" + key.ToString() + ") was not found.");
#endregion
}
if (this._keysByArrivalOrder.ContainsKey(entry.ArrivalOrder))
{
this._keysByArrivalOrder.Remove(entry.ArrivalOrder);
}
else
{
#region Report internal corruption
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Internal collection corruption detected in "
+ "OrderedHashList.Remove. Arrival order " + entry.ArrivalOrder.ToString()
+ " was not found (for key " + key.ToString() + ").");
#endregion
}
this._deletedKeysInArrivalOrder.Add(entry.Key, true);
}
#region catch ...
catch (Exception ex)
{
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Unexpected exception in OrderedHashList.Remove "
+ "removing entry from internal structures: " + ex.Message, ex);
}
#endregion
}
else
{
#region Report internal corruption
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Internal collection corruption detected in "
+ "OrderedHashList.Remove. Key (" + key.ToString() + ") was found, but "
+ "the associated entry was null.");
#endregion
}
}
}
}
#endregion
#region ContainsKey
/// <summary>Determines whether the key is in the OrderedHashList.
/// This is an O(1) operation.</summary>
/// <param name="key"></param>
/// <returns>true if the key can be found in the collection.</returns>
public bool ContainsKey(K key)
{
lock (this._lock)
{
return this._entriesByKey.ContainsKey(key);
}
}
#endregion
#region GetValueAtArrivalOrder
/// <summary>Returns only the "value" portion of the OrderedHashListEntry at the
/// specified ordinal position. This is an O(1) operation.</summary>
/// <param name="arrivalOrder">The arrival order associated with an entry.</param>
/// <returns>The value with the specified arrival order, or null/zero if that
/// arrival order doesn't exist in the collection.</returns>
public V GetValueAtArrivalOrder(int arrivalOrder)
{
lock (this._lock)
{
if (this._keysByArrivalOrder.ContainsKey(arrivalOrder))
{
K key = (K)this._keysByArrivalOrder[arrivalOrder];
if (this._entriesByKey.ContainsKey(key))
{
OrderedHashListEntry entry = (OrderedHashListEntry)this._entriesByKey[key];
if (entry != null)
{
return entry.Value;
}
else
{
#region Report internal corruption
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Internal collection corruption detected in "
+ "OrderedHashList.GetValueAtOrder. Located entry (for arrival order "
+ arrivalOrder.ToString() + " -> key " + key.ToString() + ") was null.");
#endregion
}
}
else
{
#region Report internal corruption
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Internal collection corruption detected in "
+"OrderedHashList.GetValueAtOrder. Arrival order (" + arrivalOrder.ToString() + ") was "
+"found, but the associated key was not.");
#endregion
}
}
else
{
return default(V);
}
}
}
#endregion
#region GetKeyAtArrivalOrder
/// <summary>Returns the "key" portion of the OrderedHashListEntry at the
/// specified arrival order position. This is an O(1) operation.</summary>
/// <param name="arrivalOrder">The arrival order associated with an entry.</param>
/// <returns>The key with the specified arrival order, or null/zero if that order
/// doesn't exist in the collection.</returns>
public K GetKeyAtArrivalOrder(int arrivalOrder)
{
lock (this._lock)
{
if (this._keysByArrivalOrder.ContainsKey(arrivalOrder))
{
return (K)this._keysByArrivalOrder[arrivalOrder];
}
else
{
return default(K);
}
}
}
#endregion
#region GetArrivalOrderOfKey
/// <summary>Returns the arrival order portion of OrderedHashListEntry at the
/// specified ordinal position. This is an O(1) operation.</summary>
/// <param name="key">The key associated with an entry.</param>
/// <returns>The arrival order of the specified key, or -1 if that order
/// doesn't exist in the collection.</returns>
public int GetArrivalOrderOfKey(K key)
{
lock (this._lock)
{
if (this._entriesByKey.ContainsKey(key))
{
OrderedHashListEntry entry = (OrderedHashListEntry)this._entriesByKey[key];
if (entry != null)
{
return entry.ArrivalOrder;
}
else
{
#region Report internal corruption
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Internal collection corruption detected in "
+ "OrderedHashList.GetValueAtArrivalOrder. Located entry (for key "
+ key.ToString() + ") was null.");
#endregion
}
}
else
{
return -1;
}
}
}
#endregion
#region GetKeys
/// <summary>Returns a list of keys that have arrival orders between to numbers.
/// This is an O(log n) operation.</summary>
/// <param name="fromArrivalOrder">The arrival order to start from (inclusive).</param>
/// <param name="toArrivalOrder">The arrival order to end at (inclusive).</param>
/// <returns>A List of keys whose order values fall within the given range.</returns>
public List<K> GetKeys(int fromArrivalOrder, int toArrivalOrder)
{
List<K> keys = new List<K>();
lock (this._lock)
{
try
{
if (this._entriesByKey.Count > 0)
{
int fromIndex;
int toIndex;
#region Find approximate start and end (may not be exact)
try
{
fromIndex = Math.Max(0, this.IndexOfArrivalOrder(fromArrivalOrder, true) - 1);
toIndex = Math.Min(this.IndexOfArrivalOrder(toArrivalOrder, true) + 1,
this._keysInArrivalOrder.Count - 1);
}
#region catch ...
catch (Exception ex)
{
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Unexpected exception in OrderedHashList.GetKeys "
+ "determining start and end positions to retrieve: " + ex.Message, ex);
}
#endregion
#endregion
#region Refine start and end indexes to be inclusive
try
{
bool indexReady = false;
while ((!indexReady) && (fromIndex < this._keysInArrivalOrder.Count))
{
K key = this._keysInArrivalOrder[fromIndex];
OrderedHashListEntry entry = (OrderedHashListEntry)this._entriesByKey[key];
if (entry.ArrivalOrder >= fromArrivalOrder)
{
indexReady = true;
}
else
{
fromIndex++;
}
}
indexReady = false;
while ((!indexReady) && (toIndex >= 0))
{
K key = this._keysInArrivalOrder[toIndex];
OrderedHashListEntry entry = (OrderedHashListEntry)this._entriesByKey[key];
if (entry.ArrivalOrder <= toArrivalOrder)
{
indexReady = true;
}
else
{
toIndex--;
}
}
toIndex = Math.Max(toIndex, fromIndex);
}
#region catch ...
catch (Exception ex)
{
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Unexpected exception in OrderedHashList.GetKeys "
+ "refining start and end positions to retrieve: " + ex.Message, ex);
}
#endregion
#endregion
#region Retrieve set
try
{
if (fromIndex < this._keysInArrivalOrder.Count)
{
keys.AddRange(this._keysInArrivalOrder.GetRange(fromIndex, toIndex - fromIndex + 1));
}
}
#region catch ...
catch (Exception ex)
{
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Unexpected exception in OrderedHashList.GetKeys "
+ "retrieving keyset: " + ex.Message, ex);
}
#endregion
#endregion
}
}
#region catch ...
catch (Exception ex)
{
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw ex;
}
#endregion
}
return keys;
}
#endregion
#region Indexer
/// <summary>Indexer that returns the value of the OrderedHashListEntry
/// specified by the given key. This is an O(1) operation.</summary>
/// <param name="key"></param>
/// <returns></returns>
public V this[K key]
{
get
{
lock (this._lock)
{
if (this._entriesByKey.ContainsKey(key))
{
OrderedHashListEntry entry = (OrderedHashListEntry)this._entriesByKey[key];
if (entry != null)
{
return entry.Value;
}
else
{
#region Report internal corruption
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Internal collection corruption detected in "
+ "OrderedHashList indexer. Key (" + key.ToString() + ") was found, but "
+ "the associated entry was null.");
#endregion
}
}
else
{
return default(V);
}
}
}
}
#endregion
#endregion
#region Private Helpers
#region IndexOfArrivalOrder (and helper)
private int IndexOfArrivalOrder(int arrivalOrder, bool closeEnough)
{
// already locked by caller
if ((arrivalOrder == 0) && (closeEnough))
{
return 0;
}
else if (this._keysInArrivalOrder.Count == 0)
{
return -1;
}
else
{
this.PurgeDeletedKeysInArrivalOrder();
int depth = 0;
int index = -1;
index = this.IndexOfArrivalOrder_Helper(depth, arrivalOrder, closeEnough, 0,
this._keysInArrivalOrder.Count - 1);
return index;
}
}
private int IndexOfArrivalOrder_Helper(int depth, int arrivalOrder, bool closeEnough,
int fromIndex, int toIndex)
{
// already locked by caller
#region Detect stack overflows
if (depth > 32)
{
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Stack overflow condition detected in "
+ "OrderedHashList.IndexOfArrivalOrder_Helper.");
}
#endregion
int testIndex = (fromIndex + toIndex) / 2;
K testKey;
try
{
testKey = this._keysInArrivalOrder[testIndex];
}
#region catch ...
catch (Exception ex)
{
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("OrderedHashList.IndexOfArrivalOrder_Helper had an exception "
+ "retrieving index " + testIndex.ToString() + " from key list (count: "
+is._keysInArrivalOrder.Count.ToString() + "): " + ex.Message, ex);
}
#endregion
if (this._entriesByKey.ContainsKey(testKey))
{
OrderedHashListEntry entry = (OrderedHashListEntry)this._entriesByKey[testKey];
if (entry != null)
{
if (entry.ArrivalOrder == arrivalOrder)
{
return testIndex;
}
else if (fromIndex == toIndex)
{
if (closeEnough)
{
return fromIndex;
}
else
{
return -1;
}
}
else
{
if (entry.ArrivalOrder > arrivalOrder)
{
return this.IndexOfArrivalOrder_Helper(depth + 1, arrivalOrder, closeEnough,
fromIndex, Math.Max(fromIndex, testIndex - 1));
}
else
{
return this.IndexOfArrivalOrder_Helper(depth + 1, arrivalOrder, closeEnough,
Math.Min(testIndex + 1, toIndex), toIndex);
}
}
}
else
#region Report internal corruption
{
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Internal collection corruption detected in "
+ "OrderedHashList.IndexOfArrivalOrder_Helper. Key (" + testKey.ToString()
+ ") was found, but the associated entry was null.");
}
#endregion
}
else
{
#region Report internal corruption
bool brute = false;
foreach (DictionaryEntry entry in this._entriesByKey)
{
if (entry.Key.Equals(testKey))
{
brute = true;
break;
}
}
bool second = this._entriesByKey.ContainsKey(testKey);
if (System.Diagnostics.Debugger.IsAttached)
{
System.Diagnostics.Debugger.Break();
}
throw new ApplicationException("Internal collection corruption detected in "
+ "OrderedHashList.IndexOfArrivalOrder_Helper. No entry for key (" + testKey.ToString() + ")."
+ "brute: " + brute.ToString() + ", second: " + second.ToString());
#endregion
}
}
#endregion
#region PurgeDeletedKeysInArrivalOrder
private void PurgeDeletedKeysInArrivalOrder()
{
// Already locked by caller
if (this._deletedKeysInArrivalOrder.Count > 0)
{
List<K> purgedKeysInOrder = new List<K>();
for (int index = 0; index < this._keysInArrivalOrder.Count; index++)
{
if (!this._deletedKeysInArrivalOrder.ContainsKey(this._keysInArrivalOrder[index]))
{
purgedKeysInOrder.Add(this._keysInArrivalOrder[index]);
}
}
this._keysInArrivalOrder = purgedKeysInOrder;
this._deletedKeysInArrivalOrder.Clear();
}
}
#endregion
#endregion
}
#endregionFor the latest source of this construct, visit the Kimball Method Slowly Changing Dimension component on CodePlex, and retrieve the OrderedHashList.cs file from the source.
No comments:
Post a Comment