spark/src/DataStreamSuite/DataStream/Primitives/Map/MapStream.cs

using bts;

using Bts.DataStreamCore;

using Spark.Core.Diagnostics;

using System;

namespace BCG {


   /// <summary>
   /// Provides a live Map-Stream, ie maps a IBlueRowCollection to another. The Mapping function
   /// is provided by the user, in the forst of the interface IStreamMapper and maps a row to the 
   /// desired row ie newRow = f(row), where f is the mapping fucntion. Note that the id 
   /// is preserved across the rows, ie row["id"] = f(row)["id"]
   /// </summary>
   [DataStreamUserRespresentation(false)]
   public class MapStream : MethodDrivenStream {
      protected IStreamMapper? myMapper;
      protected IChangeableStreamMapper? myChangeableMapper;
      private bool myUseNewMap = false;
      IDisposable? subscription;

      public class FuncWrapper : IStreamMapper {
         private Func<IBlueTableSchema, IBlueTableSchema> _schema_map_func;
         Func<IBlueRow, IBlueRow> _row_map_func;

         public FuncWrapper(Func<IBlueTableSchema, IBlueTableSchema> schema_map_func,
         Func<IBlueRow, IBlueRow> row_map_func) {
            _schema_map_func = schema_map_func;
            _row_map_func = row_map_func;
         }
         public IBlueTableSchema getOutputSchema(IBlueTableSchema input_schema) {
            return _schema_map_func(input_schema);
         }
         public IBlueRow map(IBlueRow row) {
            return _row_map_func(row);
         }

      }

      public MapStream(ViewStream vs, IStreamMapper mapper) : base(vs) {
         if (myUseNewMap)
            mySetRealSource(new SuperMapStream(vs, mapper));
         else
            myMapper = mapper;
      }

      public MapStream(ViewStream vs, Func<IBlueTableSchema, IBlueTableSchema> schema_map_func,
                                       Func<IBlueRow, IBlueRow> row_map_func)
                                       :
                                       this(vs, new FuncWrapper(schema_map_func, row_map_func)) {
      }

      public MapStream(ViewStream vs, IChangeableStreamMapper mapper) : this(vs, (IStreamMapper)mapper) {
         if (myUseNewMap)
            mySetRealSource(new SuperMapStream(vs, mapper));
         else {
            mapper.onMapperChange += new MapperChangeDelegate(myMapperChanged);
            myChangeableMapper = mapper;
         }
      }

      /// <summary>
      /// Gets or sets the value of the mapper. Setting the value will result in re
      /// </summary>
      private void myMapperChanged() {
         if (myLastArrivedPacket != null)
            myChangeHandler(new DataPacket(myLastArrivedPacket.newRows,
                                             myLastArrivedPacket.newRows,
                                             myLastArrivedPacket.newRows));
      }


      protected override void myHandleFirstCall() {
         ArgumentNullException.ThrowIfNull(myMapper);
         IBlueTableSchema outputSchema = myMapper.getOutputSchema(myLastArrivedPacket.newRows.schema);
         myLastFiredPacket = new DataPacket();
         myLastFiredPacket.newRows = new BlueTable(outputSchema);
         myLastFiredPacket.deletedRows = new BlueTable(outputSchema);
         foreach (IBlueRow row in myLastArrivedPacket.newRows)
            ((BlueTable)(myLastFiredPacket.newRows)).addRow(myMapper.map(row));
         myLastFiredPacket.addedRows = new BlueTable(myLastFiredPacket.newRows);
      }

      protected override void myHandleIncrementalAdd() {
         if (myLastArrivedPacket.addedRows.Count == 0) {
            Log.Gen.warn("MapStream.myHandleIncrementalAdd called with 0 addedRows");
            return;
         }
         ArgumentNullException.ThrowIfNull(myMapper);
         var addedRows = new BlueTable(myLastFiredPacket.newRows.schema);
         var newRows = ExceptionAssert.Cast(myLastFiredPacket.newRows).To<BlueTable>();
         foreach (IBlueRow row in myLastArrivedPacket.addedRows) {
            var newRow = myMapper.map(row);
            newRows.addRow(newRow);
            addedRows.addRow(newRow);
         }
         myLastFiredPacket.addedRows = addedRows;
      }

      protected override void myHandleIncrementalDelete() {
         if (myLastArrivedPacket.deletedRows.Count == 0) {
            Log.Gen.warn("MapStream.myHandleIncrementalDelete called with 0 deletedRows");
            return;
         }
         var deletedRows = new BlueTable(myLastFiredPacket.newRows.schema);
         var newRows = ExceptionAssert.Cast(myLastFiredPacket.newRows).To<BlueTable>();
         foreach (IBlueRow row in myLastArrivedPacket.deletedRows) {
            var delRow = newRows[row.Id()];
            deletedRows.addRow(delRow);
            newRows.deleteRow(row);
         }
         myLastFiredPacket.deletedRows = deletedRows;
      }


      protected override void myOnFirstSubscriber() {
         subscription = myStream.Subscribe(myChangeHandler);
      }

      protected override void myOnLastSubscriber() {
         myInit();
         subscription?.Dispose();
         subscription = null;
      }

      protected override void myOnTableClear() {
         myLastFiredPacket.deletedRows = myLastFiredPacket.newRows;
         myLastFiredPacket.newRows = new BlueTable(myLastFiredPacket.deletedRows.schema);
         myLastFiredPacket.addedRows = new BlueTable(myLastFiredPacket.deletedRows.schema);
      }


   }

   /*
   public class DistinctColumnSplitterStream : ViewStream
   {
      private string _split_col;
      private ViewStream _stream;

      public DistinctColumnSplitterStream(ViewStream vs, string split_col)
      {
         _split_col = split_col;
         _stream = vs;
      }

      protected override void myOnFirstSubscriber()
      {
         _stream.onChange += _stream_onChange;
      }

      protected override void myOnLastSubscriber()
      {
         _stream.onChange -= _stream_onChange;
      }

      private void _stream_onChange(object newValue)
      {
         DataPacket dp = newValue as DataPacket;
         System.Collections.Generic.Dictionary<object, BlueTable> dict = new System.Collections.Generic.Dictionary<object, BlueTable>();
         if (dp != null)
         {
            foreach (IBlueRow r in dp.newRows)
            {
               object split_val = r[_split_col];
               if (!dict.ContainsKey(split_val))
                  dict.Add(split_val, _make_table(split_val, r.schema.columnInfo));
               add_row(r, dict[split_val]);
            }
         }
      }

      private IBlueTableSchema add_id_col(BlueTable[] without_ids)
      {
         //ColumnInfo[] ci = new ColumnInfo[without_id.columnInfo.Length + 1];
         //Array.Copy(without_id.columnInfo, 0, ci, 1, without_id.columnInfo.Length);
         //ci[0] = new ColumnInfo("id", 0, typeof(int));
         System.Collections.Generic.List<ColumnInfo> cols = new System.Collections.Generic.List<ColumnInfo>();
         int place_holder = 0;
         foreach (BlueTable bt in without_ids)
            for (int i = 0; i < bt.schema.columnInfo.Length; i++)
               cols.Add(new ColumnInfo(bt.schema.columnNames[i], place_holder++, bt.schema.columnInfo[i].colType);
         ColumnInfo[] res_ci = new ColumnInfo[cols.Count + 1];
      }

      private BlueTable _make_aggregate_table(System.Collections.Generic.Dictionary<object, BlueTable> dict)
      {
         int size = 0;
         foreach (System.Collections.Generic.KeyValuePair<object, BlueTable> kvp in dict)
         {
            if (kvp.Value.Count > size)
               size = kvp.Value.Count;
         }
         BlueTable res = new BlueTable
         for (int i = 0; i < size; i++)
         {

         }
      }

      private BlueTable _make_table(object split_col_val, ColumnInfo[] base_ci)
      {
         string split_col_prefix = String.Format("{0}-{1}:", _split_col, split_col_val.ToString());
         ColumnInfo[] split_ci = new ColumnInfo[base_ci.Length - 1];
         for (int i = 0; i < base_ci.Length; i++)
         {
            if(base_ci[i].colName != "id")
               split_ci[i] = new ColumnInfo(split_col_prefix + base_ci[i].colName, i, base_ci[i].colType);
         }
         return new BlueTable("", split_ci);
      }

      private void add_row(IBlueRow r, BlueTable t)
      {
         object[] vals = new object[r.schema.columnInfo.Length - 1];
         int corrector = 0;
         for (int i = 0; i < vals.Length; i++)
         {
            if (r.schema.columnNames[i] == "id")
               corrector = 1;
            vals[i] = r[i + corrector];
         }
         t.addRow(new BlueRow(t.schema, vals));
      }
   }*/
}