Files
spark/src/DataStreamSuite/DataStreamCore/DataPacket.cs
T

322 lines
9.8 KiB
C#

using System;
using System.Collections;
using System.Data;
using BCG.Utilities.General;
namespace BCG
{
[Serializable]
public class DataPacket : ICloneable, IDataPacket
{
public class BlueRowIdComparator
{
private string[] myIdColumns = new string[]{"id"};
public BlueRowIdComparator(string[] idCols)
{
if ( idCols.Length != 0 )
{
myIdColumns = idCols;
}
}
public bool blueRowIdEqualEvaluator(object a, object b)
{
if ( a == null || b == null )
{
return false;
}
if ( !(a is IBlueRow) || !(b is IBlueRow) )
{
return false;
}
IBlueRow lhs = (IBlueRow)a;
IBlueRow rhs = (IBlueRow)b;
bool result = true;
foreach(string idColumn in myIdColumns)
{
result = result && lhs[idColumn].Equals(rhs[idColumn]);
if ( !result )
break;
}
return result;
}
}
public IBlueRowCollection newRows, addedRows, deletedRows;
public DataPacket () {}
public DataPacket (IBlueRowCollection newRows, IBlueRowCollection addedRows, IBlueRowCollection deletedRows)
{
this.newRows = newRows;
this.addedRows = addedRows;
this.deletedRows = deletedRows;
}
public void get_new_added_deleted(out ICollection new_rows, out ICollection added_rows, out ICollection deleted_rows)
{
new_rows = newRows;
added_rows = addedRows;
deleted_rows = deletedRows;
}
public IBlueRowCollection getTrulyAddedRows(params string[] idColumns)
{
BlueTable truly = new BlueTable(addedRows.schema);
BlueRowIdComparator c = new BlueRowIdComparator(idColumns);
CollectionHelpers.EqualDelegate ed =
new CollectionHelpers.EqualDelegate(c.blueRowIdEqualEvaluator);
ICollection ta = CollectionHelpers.difference(addedRows, deletedRows, ed);
foreach(IBlueRow r in ta)
{
truly.addRow(r);
}
return truly;
}
public virtual IBlueRowCollection getTrulyAddedRows ()
{
BlueTable retVal = new BlueTable (schema);
foreach (IBlueRow r in addedRows)
if (!deletedRows.containsID ((int)r["id"]))
retVal.addRow (r);
return retVal;
}
public virtual IBlueRowCollection getTrulyDeletedRows ()
{
BlueTable retVal = new BlueTable (schema);
foreach (IBlueRow r in deletedRows)
if (!addedRows.containsID ((int)r["id"]))
retVal.addRow (r);
return retVal;
}
public virtual IBlueRowCollection getTrulyChangedRows ()
{
BlueTable retVal = new BlueTable (schema);
foreach (IBlueRow r in addedRows)
if (deletedRows.containsID ((int)r["id"]))
retVal.addRow (r);
return retVal;
}
public IBlueRowCollection getTrulyDeletedRows(params string[] idColumns)
{
BlueTable truly = new BlueTable(deletedRows.schema);
BlueRowIdComparator c = new BlueRowIdComparator(idColumns);
CollectionHelpers.EqualDelegate ed =
new CollectionHelpers.EqualDelegate(c.blueRowIdEqualEvaluator);
ICollection td = CollectionHelpers.difference(deletedRows, addedRows, ed);
foreach(IBlueRow r in td)
{
truly.addRow(r);
}
return truly;
}
public IBlueRowCollection getTrulyChangedRows(params string[] idColumns)
{
BlueTable truly = new BlueTable(addedRows.schema);
BlueRowIdComparator c = new BlueRowIdComparator(idColumns);
CollectionHelpers.EqualDelegate ed =
new CollectionHelpers.EqualDelegate(c.blueRowIdEqualEvaluator);
ICollection ta = CollectionHelpers.difference(addedRows, deletedRows, ed);
ICollection tc = CollectionHelpers.difference(addedRows, ta, ed);
foreach(IBlueRow r in tc)
{
truly.addRow(r);
}
return truly;
}
public IBlueRowCollection getTrulyChangedOldRows(params string[] idColumns)
{
BlueTable truly = new BlueTable(deletedRows.schema);
BlueRowIdComparator c = new BlueRowIdComparator(idColumns);
CollectionHelpers.EqualDelegate ed =
new CollectionHelpers.EqualDelegate(c.blueRowIdEqualEvaluator);
ICollection td = CollectionHelpers.difference(deletedRows, addedRows, ed);
ICollection tco = CollectionHelpers.difference(deletedRows, td, ed);
foreach(IBlueRow r in tco)
{
truly.addRow(r);
}
return truly;
}
public TablePacket toTablePacket ()
{
return new TablePacket (BlueDataUtility.toDataTable (newRows), BlueDataUtility.toDataTable (addedRows),
BlueDataUtility.toDataTable (deletedRows));
}
public IBlueTableSchema schema
{
get
{
IBlueTableSchema retVal = newRows.schema != null ? newRows.schema :
addedRows.schema != null ? addedRows.schema : deletedRows.schema;
return retVal;
}
}
/*
status D total A total D this A This D' total A' Total
x 0 0 0 0 0 0
0 0 0 1 1 this+
0 0 1 0 1this
0 0 1 1 1 this 1 this+
x 0 1 0 0
error 0 1 0 1
0 1 1 0 1 total-
0 1 1 1 1 total-, 1 this +
x 1 0 0 0
1 0 0 1 1 this+
error 1 0 1 0
error 1 0 1 1
x 1 1 0 0
error 1 1 0 1
1 1 1 0 1 total -1 keep the old delete and remove the old add
1 1 1 1 1 total-, 1 this +
d' = dthis - (dthis intersect a) + d // keep the old delete
a' = a - (dthis intersect a) + athis // keep the new add
* */
public static DataPacket combineWithIds(ICollection packets, string[] idColumns)
{
if ( packets.Count == 0 )
{
return new DataPacket(
new BlueTable(new DataTable()),
new BlueTable(new DataTable()),
new BlueTable(new DataTable()));
}
DataPacket.BlueRowIdComparator c = new DataPacket.BlueRowIdComparator(idColumns);
CollectionHelpers.EqualDelegate ed =
new CollectionHelpers.EqualDelegate(c.blueRowIdEqualEvaluator);
ArrayList added = new ArrayList();
ArrayList deleted = new ArrayList();
BlueTable dt = null;
BlueTable at = null;
IBlueRowCollection nt = null;
foreach(DataPacket dp in packets)
{
ArrayList oldDeleted = deleted;
ArrayList oldAdded = added;
added = new ArrayList();
deleted = new ArrayList();
if ( dt == null )
{
dt = new BlueTable(dp.schema);
at = new BlueTable(dp.schema);
}
deleted.AddRange(oldDeleted);
deleted.AddRange( CollectionHelpers.difference(
CollectionHelpers.difference(dp.deletedRows, oldAdded, ed),
oldDeleted, ed) );
added.AddRange(dp.addedRows);
added.AddRange( CollectionHelpers.difference(
CollectionHelpers.difference(oldAdded, dp.deletedRows, ed),
added, ed) );
nt = dp.newRows;
}
foreach ( IBlueRow dr in added )
{
at.addRow(dr);
}
foreach ( IBlueRow dr in deleted )
{
dt.addRow(dr);
}
return new DataPacket(nt,at,dt);
}
public DataPacket deepCopy ()
{
return new DataPacket (new OrderedBlueTable (newRows),
new OrderedBlueTable (newRows),
new BlueTable (deletedRows));
}
public virtual DataPacket createImmutable ()
{
return
new DataPacket(
newRows is IImmutableRowCollection ? newRows : new ImmutableBlueTable(newRows),
addedRows,
deletedRows);
}
#region ICloneable Members
public object Clone()
{
return MemberwiseClone();
}
#endregion
public string toIncrementalIDString()
{
System.Text.StringBuilder sb = new System.Text.StringBuilder(schema.tableName);
sb.Append ("\ndeleted : ");
foreach (IBlueRow r in deletedRows)
{
sb.Append(r["id"]);
sb.Append(',');
}
sb.Append("\nadded : ");
foreach (IBlueRow r in addedRows)
{
sb.Append(r["id"]);
sb.Append(',');
}
sb.Append("\n");
return sb.ToString();
}
public override string ToString()
{
return String.Format("DataPacket, schema {0} n {1} d {2} a {3}", schema.tableName, newRows.Count, deletedRows.Count, addedRows.Count);
}
}
public static class DataPacketExtensions {
public static bool HasChanges(this DataPacket dataPacket) =>
dataPacket.addedRows.Count > 0 || dataPacket.deletedRows.Count > 0;
}
}