Prediction engine now uses IRowToRowMapper

Author: TomFinley

src/Microsoft.ML.Api/ComponentCreation.cs

@@ -5,6 +5,7 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using Microsoft.ML.Core.Data;
 using Microsoft.ML.Runtime.CommandLine;
 using Microsoft.ML.Runtime.Data;
 using Microsoft.ML.Runtime.Model;
@@ -189,6 +190,26 @@ public static PredictionEngine<TSrc, TDst> CreatePredictionEngine<TSrc, TDst>(th
             return new PredictionEngine<TSrc, TDst>(env, dataPipe, ignoreMissingColumns, inputSchemaDefinition, outputSchemaDefinition);
         }
 
+        /// <summary>
+        /// Create an on-demand prediction engine.
+        /// </summary>
+        /// <param name="env">The host environment to use.</param>
+        /// <param name="transformer">The transformer.</param>
+        /// <param name="ignoreMissingColumns">Whether to ignore missing columns in the data view.</param>
+        /// <param name="inputSchemaDefinition">The optional input schema. If <c>null</c>, the schema is inferred from the <typeparamref name="TSrc"/> type.</param>
+        /// <param name="outputSchemaDefinition">The optional output schema. If <c>null</c>, the schema is inferred from the <typeparamref name="TDst"/> type.</param>
+        public static PredictionEngine<TSrc, TDst> CreatePredictionEngine<TSrc, TDst>(this IHostEnvironment env, ITransformer transformer,
+            bool ignoreMissingColumns = false, SchemaDefinition inputSchemaDefinition = null, SchemaDefinition outputSchemaDefinition = null)
+            where TSrc : class
+            where TDst : class, new()
+        {
+            Contracts.CheckValue(env, nameof(env));
+            env.CheckValue(transformer, nameof(transformer));
+            env.CheckValueOrNull(inputSchemaDefinition);
+            env.CheckValueOrNull(outputSchemaDefinition);
+            return new PredictionEngine<TSrc, TDst>(env, transformer, ignoreMissingColumns, inputSchemaDefinition, outputSchemaDefinition);
+        }
+
         /// <summary>
         /// Create a prediction engine.
         /// This encapsulates the 'classic' prediction problem, where the input is denoted by the float array of features,

src/Microsoft.ML.Api/DataViewConstructionUtils.cs

@@ -24,8 +24,6 @@ internal sealed class MapTransform<TSrc, TDst> : LambdaTransformBase, ITransform
         where TDst : class, new()
     {
         private const string RegistrationNameTemplate = "MapTransform<{0}, {1}>";
-
-        private readonly IDataView _source;
         private readonly Action<TSrc, TDst> _mapAction;
         private readonly MergedSchema _schema;
 
@@ -56,20 +54,20 @@ public MapTransform(IHostEnvironment env, IDataView source, Action<TSrc, TDst> m
             SchemaDefinition inputSchemaDefinition = null, SchemaDefinition outputSchemaDefinition = null)
             : base(env, RegistrationName, saveAction, loadFunc)
         {
-            Host.AssertValue(source, "source");
-            Host.AssertValue(mapAction, "mapAction");
+            Host.AssertValue(source);
+            Host.AssertValue(mapAction);
             Host.AssertValueOrNull(inputSchemaDefinition);
             Host.AssertValueOrNull(outputSchemaDefinition);
 
-            _source = source;
+            Source = source;
             _mapAction = mapAction;
             _inputSchemaDefinition = inputSchemaDefinition;
             _typedSource = TypedCursorable<TSrc>.Create(Host, Source, false, inputSchemaDefinition);
             var outSchema = outputSchemaDefinition == null
                ? InternalSchemaDefinition.Create(typeof(TDst), SchemaDefinition.Direction.Write)
                : InternalSchemaDefinition.Create(typeof(TDst), outputSchemaDefinition);
 
-            _schema = MergedSchema.Create(_source.Schema, outSchema);
+            _schema = MergedSchema.Create(Source.Schema, outSchema);
         }
 
         /// <summary>
@@ -81,26 +79,20 @@ private MapTransform(IHostEnvironment env, MapTransform<TSrc, TDst> transform, I
             Host.AssertValue(transform);
             Host.AssertValue(newSource);
 
-            _source = newSource;
+            Source = newSource;
             _mapAction = transform._mapAction;
             _typedSource = TypedCursorable<TSrc>.Create(Host, newSource, false, transform._inputSchemaDefinition);
 
             _schema = MergedSchema.Create(newSource.Schema, transform._schema.AddedSchema);
         }
 
-        public bool CanShuffle
-        {
-            get { return _source.CanShuffle; }
-        }
+        public bool CanShuffle => Source.CanShuffle;
 
-        public ISchema Schema
-        {
-            get { return _schema; }
-        }
+        public ISchema Schema => _schema;
 
         public long? GetRowCount(bool lazy = true)
         {
-            return _source.GetRowCount(lazy);
+            return Source.GetRowCount(lazy);
         }
 
         public IRowCursor GetRowCursor(Func<int, bool> predicate, IRandom rand = null)
@@ -136,10 +128,7 @@ public IRowCursor[] GetRowCursorSet(out IRowCursorConsolidator consolidator, Fun
             return cursors;
         }
 
-        public IDataView Source
-        {
-            get { return _source; }
-        }
+        public IDataView Source { get; }
 
         public IDataTransform ApplyToData(IHostEnvironment env, IDataView newSource)
         {
@@ -161,13 +150,13 @@ public Func<int, bool> GetDependencies(Func<int, bool> predicate)
             return _typedSource.GetDependencies(srcPredicate);
         }
 
-        ISchema IRowToRowMapper.InputSchema => _source.Schema;
+        ISchema IRowToRowMapper.InputSchema => Source.Schema;
 
         public IRow GetRow(IRow input, Func<int, bool> active, out Action disposer)
         {
             Host.CheckValue(input, nameof(input));
             Host.CheckValue(active, nameof(active));
-            Host.CheckParam(input.Schema == _source.Schema, nameof(input), "Schema of input row must be the same as the schema the mapper is bound to");
+            Host.CheckParam(input.Schema == Source.Schema, nameof(input), "Schema of input row must be the same as the schema the mapper is bound to");
 
             var src = new TSrc();
             var dst = new TDst();
@@ -234,7 +223,7 @@ public ValueGetter<TValue> GetGetter<TValue>(int col)
                     () =>
                     {
                         if (!IsGood)
-                            throw Contracts.Except("Getter is called when the cursor is {0}, which is not allowed.", Input.State);
+                            throw Ch.Except("Getter is called when the cursor is {0}, which is not allowed.", Input.State);
                     };
                 return _row.GetGetterCore<TValue>(col, isGood);
             }
@@ -253,7 +242,6 @@ public override void Dispose()
 
         private sealed class Row : IRow
         {
-            private readonly ISchema _schema;
             private readonly IRow<TSrc> _input;
             private readonly IRow _appendedRow;
             private readonly bool[] _active;
@@ -269,15 +257,15 @@ private sealed class Row : IRow
 
             public long Position => _input.Position;
 
-            public ISchema Schema => _schema;
+            public ISchema Schema { get; }
 
             public Row(IRow<TSrc> input, MapTransform<TSrc, TDst> parent, Func<int, bool> active, TSrc src, TDst dst)
             {
                 _input = input;
                 _parent = parent;
-                _schema = parent.Schema;
+                Schema = parent.Schema;
 
-                _active = Utils.BuildArray(_schema.ColumnCount, active);
+                _active = Utils.BuildArray(Schema.ColumnCount, active);
                 _src = src;
                 _dst = dst;
 
@@ -321,7 +309,7 @@ public ValueGetter<UInt128> GetIdGetter()
 
             public bool IsColumnActive(int col)
             {
-                _parent.Host.Check(0 <= col && col < _schema.ColumnCount);
+                _parent.Host.Check(0 <= col && col < Schema.ColumnCount);
                 return _active[col];
             }
         }

src/Microsoft.ML.Api/MapTransform.cs

@@ -8,6 +8,8 @@
 using System.IO;
 using Microsoft.ML.Runtime.Data;
 using Microsoft.ML.Runtime.Model;
+using Microsoft.ML.Core.Data;
+using System;
 
 namespace Microsoft.ML.Runtime.Api
 {
@@ -38,21 +40,7 @@ internal BatchPredictionEngine(IHostEnvironment env, Stream modelStream, bool ig
 
             // Initialize pipe.
             _srcDataView = DataViewConstructionUtils.CreateFromEnumerable(env, new TSrc[] { }, inputSchemaDefinition);
-
-            // Load transforms.
-            var pipe = env.LoadTransforms(modelStream, _srcDataView);
-
-            // Load predictor (if present) and apply default scorer.
-            // REVIEW: distinguish the case of predictor / no predictor?
-            var predictor = env.LoadPredictorOrNull(modelStream);
-            if (predictor != null)
-            {
-                var roles = ModelFileUtils.LoadRoleMappingsOrNull(env, modelStream);
-                pipe = roles != null
-                    ? env.CreateDefaultScorer(new RoleMappedData(pipe, roles, opt: true), predictor)
-                    : env.CreateDefaultScorer(new RoleMappedData(pipe, label: null, "Features"), predictor);
-            }
-
+            var pipe = DataViewConstructionUtils.LoadPipeWithPredictor(env, modelStream, _srcDataView);
             _pipeEngine = new PipeEngine<TDst>(env, pipe, ignoreMissingColumns, outputSchemaDefinition);
         }
 
@@ -150,24 +138,64 @@ public sealed class PredictionEngine<TSrc, TDst>
         where TSrc : class
         where TDst : class, new()
     {
-        private readonly BatchPredictionEngine<TSrc, TDst> _engine;
+        private readonly DataViewConstructionUtils.InputRow<TSrc> _inputRow;
+        private readonly IRow<TDst> _outputRow;
+        private readonly Action _disposer;
+        private TDst _result;
 
         internal PredictionEngine(IHostEnvironment env, Stream modelStream, bool ignoreMissingColumns,
             SchemaDefinition inputSchemaDefinition = null, SchemaDefinition outputSchemaDefinition = null)
+            : this(env, StreamChecker(env, modelStream), ignoreMissingColumns, inputSchemaDefinition, outputSchemaDefinition)
         {
-            Contracts.CheckValue(env, nameof(env));
-            var singleThreadedEnv = env.Register("SingleThreaded", conc: 1);
-            _engine = new BatchPredictionEngine<TSrc, TDst>(singleThreadedEnv, modelStream, ignoreMissingColumns,
-                inputSchemaDefinition, outputSchemaDefinition);
+        }
+
+        private static Func<ISchema, IRowToRowMapper> StreamChecker(IHostEnvironment env, Stream modelStream)
+        {
+            env.CheckValue(modelStream, nameof(modelStream));
+            return schema =>
+            {
+                var pipe = DataViewConstructionUtils.LoadPipeWithPredictor(env, modelStream, new EmptyDataView(env, schema));
+                var transformer = new TransformWrapper(env, pipe);
+                env.CheckParam(transformer.IsRowToRowMapper, nameof(transformer), "Must be a row to row mapper");
+                return transformer.GetRowToRowMapper(schema);
+            };
         }
 
         internal PredictionEngine(IHostEnvironment env, IDataView dataPipe, bool ignoreMissingColumns,
             SchemaDefinition inputSchemaDefinition = null, SchemaDefinition outputSchemaDefinition = null)
+            : this(env, new TransformWrapper(env, env.CheckRef(dataPipe, nameof(dataPipe))), ignoreMissingColumns, inputSchemaDefinition, outputSchemaDefinition)
+        {
+        }
+
+        internal PredictionEngine(IHostEnvironment env, ITransformer transformer, bool ignoreMissingColumns,
+            SchemaDefinition inputSchemaDefinition = null, SchemaDefinition outputSchemaDefinition = null)
+            : this(env, TransformerChecker(env, transformer), ignoreMissingColumns, inputSchemaDefinition, outputSchemaDefinition)
+        {
+        }
+
+        private static Func<ISchema, IRowToRowMapper> TransformerChecker(IExceptionContext ectx, ITransformer transformer)
+        {
+            ectx.CheckValue(transformer, nameof(transformer));
+            ectx.CheckParam(transformer.IsRowToRowMapper, nameof(transformer), "Must be a row to row mapper");
+            return transformer.GetRowToRowMapper;
+        }
+
+        private PredictionEngine(IHostEnvironment env, Func<ISchema, IRowToRowMapper> makeMapper, bool ignoreMissingColumns,
+                 SchemaDefinition inputSchemaDefinition, SchemaDefinition outputSchemaDefinition)
         {
             Contracts.CheckValue(env, nameof(env));
-            var singleThreadedEnv = env.Register("SingleThreaded", conc: 1);
-            _engine = new BatchPredictionEngine<TSrc, TDst>(singleThreadedEnv, dataPipe, ignoreMissingColumns,
-                inputSchemaDefinition, outputSchemaDefinition);
+            env.AssertValue(makeMapper);
+
+            _inputRow = DataViewConstructionUtils.CreateInputRow<TSrc>(env, inputSchemaDefinition);
+            var mapper = makeMapper(_inputRow.Schema);
+            var cursorable = TypedCursorable<TDst>.Create(env, new EmptyDataView(env, mapper.Schema), ignoreMissingColumns, outputSchemaDefinition);
+            var outputRow = mapper.GetRow(_inputRow, col => true, out _disposer);
+            _outputRow = cursorable.GetRow(outputRow);
+        }
+
+        ~PredictionEngine()
+        {
+            _disposer?.Invoke();
         }
 
         /// <summary>
@@ -178,21 +206,11 @@ internal PredictionEngine(IHostEnvironment env, IDataView dataPipe, bool ignoreM
         public TDst Predict(TSrc example)
         {
             Contracts.CheckValue(example, nameof(example));
-            int count = 0;
-            TDst result = null;
-            foreach (var item in _engine.Predict(new[] { example }, true))
-            {
-                if (count == 0)
-                    result = item;
-
-                count++;
-                if (count > 1)
-                    break;
-            }
-
-            if (count > 1)
-                throw Contracts.Except("Prediction pipeline must return at most one prediction per example. If it isn't, use BatchPredictionEngine.");
-            return result;
+            _inputRow.AcceptValues(example);
+            if (_result == null)
+                _result = new TDst();
+            _outputRow.FillValues(_result);
+            return _result;
         }
     }
 

src/Microsoft.ML.Api/PredictionEngine.cs

@@ -23,7 +23,7 @@ public PredictionFunction(IHostEnvironment env, ITransformer transformer)
             env.CheckValue(transformer, nameof(transformer));
 
             IDataView dv = env.CreateDataView(new TSrc[0]);
-            _engine = env.CreatePredictionEngine<TSrc, TDst>(transformer.Transform(dv));
+            _engine = env.CreatePredictionEngine<TSrc, TDst>(transformer);
         }
 
         public TDst Predict(TSrc example) => _engine.Predict(example);

src/Microsoft.ML.Api/PredictionFunction.cs

@@ -26,6 +26,21 @@ public interface IRow<TRow> : IRow
         void FillValues(TRow row);
     }
 
+    /// <summary>
+    /// This interface is an <see cref="IRow"/> with 'stringly typed' binding.
+    /// It can accept values of type <typeparamref name="TRow"/> and present the value as a row.
+    /// </summary>
+    /// <typeparam name="TRow"></typeparam>
+    public interface IInputRow<TRow> : IRow
+        where TRow : class
+    {
+        /// <summary>
+        /// Accepts the fields of the user-supplied <paramref name="row"/> object and publishes the instance as a row.
+        /// </summary>
+        /// <param name="row">The row object. Cannot be null.</param>
+        void AcceptValues(TRow row);
+    }
+
     /// <summary>
     /// This interface provides cursoring through a <see cref="IDataView"/> via a 'strongly typed' binding.
     /// It can populate the user-supplied object's fields with the values of the current row.
@@ -239,11 +254,11 @@ private abstract class TypedRowBase : IRow<TRow>
             private readonly IRow _input;
             private readonly Action<TRow>[] _setters;
 
-            public long Batch { get { return _input.Batch; } }
+            public long Batch => _input.Batch;
 
-            public long Position { get { return _input.Position; } }
+            public long Position => _input.Position;
 
-            public ISchema Schema { get { return _input.Schema; } }
+            public ISchema Schema => _input.Schema;
 
             public TypedRowBase(TypedCursorable<TRow> parent, IRow input, string channelMessage)
             {

src/Microsoft.ML.Api/TypedCursor.cs

@@ -30,6 +30,7 @@ public ChainedRowToRowMapper(ISchema inputSchema, IRowToRowMapper[] mappers)
             Contracts.CheckValueOrNull(mappers);
             _innerMappers = Utils.Size(mappers) > 0 ? mappers : _empty;
             InputSchema = inputSchema;
+            Schema = Utils.Size(mappers) > 0 ? mappers[mappers.Length - 1].Schema : inputSchema;
         }
 
         public Func<int, bool> GetDependencies(Func<int, bool> predicate)
@@ -67,7 +68,7 @@ public IRow GetRow(IRow input, Func<int, bool> active, out Action disposer)
             // computed based on the dependencies of the next one in the chain.
             var deps = new Func<int, bool>[_innerMappers.Length];
             deps[deps.Length - 1] = active;
-            for (int i = deps.Length - 1; i <= 1; --i)
+            for (int i = deps.Length - 1; i >= 1; --i)
                 deps[i - 1] = _innerMappers[i].GetDependencies(deps[i]);
 
             IRow result = input;

src/Microsoft.ML.Data/DataView/ChainedRowToRowMapper.cs

@@ -93,6 +93,8 @@ protected PredictionTransformerBase(IHost host, ModelLoadContext ctx)
         /// <returns>The transformed <see cref="IDataView"/></returns>
         public abstract IDataView Transform(IDataView input);
 
+        public abstract IRowToRowMapper GetRowToRowMapper(ISchema inputSchema);
+
         protected void SaveModel(ModelSaveContext ctx)
         {
             // *** Binary format ***
@@ -187,8 +189,6 @@ protected virtual void SaveCore(ModelSaveContext ctx)
             SaveModel(ctx);
             ctx.SaveStringOrNull(FeatureColumn);
         }
-
-        public abstract IRowToRowMapper GetRowToRowMapper(ISchema inputSchema);
     }
 
     /// <summary>

src/Microsoft.ML.Data/Scorers/PredictionTransformer.cs

@@ -485,7 +485,7 @@ public Func<int, bool> GetDependencies(Func<int, bool> predicate)
             {
                 for (int i = 0; i < Schema.ColumnCount; i++)
                 {
-                    if (predicate(i) && _inputSchema.Feature != null)
+                    if (predicate(i) && InputRoleMappedSchema.Feature != null)
                         return col => col == InputRoleMappedSchema.Feature.Index;
                 }
                 return col => false;