spark/src/DataStreamSuite/DataStream/Primitives/AddColumns/MathStream.cs

using Bts.ArithmeticExpressions;

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace BCG
{

   public class BooleanStringReplaceStream : ViewStream, IStreamMapper
   {
      private BooleanExpressionFilter _bool;
      private string _column, _value;
      private IBlueTableSchema _schema;
      public BooleanStringReplaceStream(ViewStream vs, string boolean_expression, string column, string value)
      {
         _column = column;
         _value = value;
         if(!string.IsNullOrEmpty(boolean_expression))
            _bool = new BooleanExpressionFilter(boolean_expression);
         mySetRealSource(new MapStream(vs, this));
      }
      public IBlueTableSchema getOutputSchema(IBlueTableSchema inputSchema)
      {
         _schema = inputSchema;
         return inputSchema;
      }
      public IBlueRow map(IBlueRow row)
      {
         MutableBlueRow result = new MutableBlueRow(_schema, row);
         if(_bool == null || _bool.filter(row))
            result.setValue(_column, _value);
         return result;
      }

   }

   public class MathStream : ViewStream, IStreamMapper
   {
      private IArithmeticProgram myExp;
      private BooleanExpressionFilter _bool;
      List<string> myExtraColumns = new List<string>();
      BlueTableSchema mySchema;
       

      public MathStream(ViewStream vs, string mathExpression)
         : this(vs, mathExpression, string.Empty)
      {
      }

      public MathStream(ViewStream vs, string mathExpression, string boolean_expression)
      {
         myExp = ArithmeticProgramParser.Parse(mathExpression) ??
            throw new Exception($"Failed to parse ArithmeticExpression '{mathExpression}'");
         myExtraColumns = myExp.OutputIdentifiers.ToList();
         if(!string.IsNullOrEmpty(boolean_expression))
            _bool = new BooleanExpressionFilter(boolean_expression);
         mySetRealSource(new MapStream(vs, this));
      }

      #region IStreamMapper Members

      public IBlueTableSchema getOutputSchema(IBlueTableSchema inputSchema)
      {
         if (mySchema == null)
         {
            Dictionary<string, ColumnInfo> colIdx = new Dictionary<string, ColumnInfo>();
            foreach (ColumnInfo ci in inputSchema.columnInfo)
               colIdx[ci.colName] = ci;

            foreach (string ncn in myExtraColumns)
            {
               ColumnInfo newC;
               if (colIdx.ContainsKey(ncn))
                  newC = new ColumnInfo(colIdx[ncn].colName, colIdx[ncn].colNo, typeof(double));
               else
                  newC = new ColumnInfo(ncn, colIdx.Count, typeof(double));
               colIdx[ncn] = newC;
            }
            mySchema = new BlueTableSchema(colIdx.Values.ToArray(), "math(" + inputSchema.tableName + ")");
         }
         return mySchema;
      }

      #endregion

      #region IDynamicStreamMapper Members

      public IBlueRow map(IBlueRow row)
      {
         MutableBlueRow result = new MutableBlueRow(mySchema, row);
         if(_bool == null || _bool.filter(row))
         {
            myExp.Evaluate(result.AsDictionary());
         }
         return result;
      }

      #endregion
   }
}