Address comments

Author: wschin

src/Microsoft.ML.Data/DataView/CompositeSchema.cs renamed to src/Microsoft.ML.Data/DataView/ZipBinding.cs

@@ -3,8 +3,6 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
-using System.Collections.Generic;
-using System.Linq;
 using Microsoft.ML.Internal.Utilities;
 
 namespace Microsoft.ML.Data

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

@@ -25,7 +25,7 @@ public sealed class ZipDataView : IDataView
 
         private readonly IHost _host;
         private readonly IDataView[] _sources;
-        private readonly ZipBinding _compositeSchema;
+        private readonly ZipBinding _zipBinding;
 
         public static IDataView Create(IHostEnvironment env, IEnumerable<IDataView> sources)
         {
@@ -47,12 +47,12 @@ private ZipDataView(IHost host, IDataView[] sources)
 
             _host.Assert(Utils.Size(sources) > 1);
             _sources = sources;
-            _compositeSchema = new ZipBinding(_sources.Select(x => x.Schema).ToArray());
+            _zipBinding = new ZipBinding(_sources.Select(x => x.Schema).ToArray());
         }
 
         public bool CanShuffle { get { return false; } }
 
-        public Schema Schema => _compositeSchema.OutputSchema;
+        public Schema Schema => _zipBinding.OutputSchema;
 
         public long? GetRowCount()
         {
@@ -75,7 +75,7 @@ public RowCursor GetRowCursor(Func<int, bool> predicate, Random rand = null)
             _host.CheckValue(predicate, nameof(predicate));
             _host.CheckValueOrNull(rand);
 
-            var srcPredicates = _compositeSchema.GetInputPredicates(predicate);
+            var srcPredicates = _zipBinding.GetInputPredicates(predicate);
 
             // REVIEW: if we know the row counts, we could only open cursor if it has needed columns, and have the
             // outer cursor handle the early stopping. If we don't know row counts, we need to open all the cursors because
@@ -106,7 +106,7 @@ public RowCursor[] GetRowCursorSet(Func<int, bool> predicate, int n, Random rand
         private sealed class Cursor : RootCursorBase
         {
             private readonly RowCursor[] _cursors;
-            private readonly ZipBinding _compositeSchema;
+            private readonly ZipBinding _zipBinding;
             private readonly bool[] _isColumnActive;
             private bool _disposed;
 
@@ -119,8 +119,8 @@ public Cursor(ZipDataView parent, RowCursor[] srcCursors, Func<int, bool> predic
                 Ch.AssertValue(predicate);
 
                 _cursors = srcCursors;
-                _compositeSchema = parent._compositeSchema;
-                _isColumnActive = Utils.BuildArray(_compositeSchema.ColumnCount, predicate);
+                _zipBinding = parent._zipBinding;
+                _isColumnActive = Utils.BuildArray(_zipBinding.ColumnCount, predicate);
             }
 
             protected override void Dispose(bool disposing)
@@ -172,19 +172,19 @@ protected override bool MoveManyCore(long count)
                 return true;
             }
 
-            public override Schema Schema => _compositeSchema.OutputSchema;
+            public override Schema Schema => _zipBinding.OutputSchema;
 
             public override bool IsColumnActive(int col)
             {
-                _compositeSchema.CheckColumnInRange(col);
+                _zipBinding.CheckColumnInRange(col);
                 return _isColumnActive[col];
             }
 
             public override ValueGetter<TValue> GetGetter<TValue>(int col)
             {
                 int dv;
                 int srcCol;
-                _compositeSchema.GetColumnSource(col, out dv, out srcCol);
+                _zipBinding.GetColumnSource(col, out dv, out srcCol);
                 return _cursors[dv].GetGetter<TValue>(srcCol);
             }
         }

src/Microsoft.ML.Transforms/GroupTransform.cs

@@ -195,14 +195,18 @@ private sealed class GroupSchema : ISchema
             private readonly IExceptionContext _ectx;
             private readonly Schema _input;
 
+            // Column names in source schema used to group rows.
             private readonly string[] _groupColumns;
+            // Column names in source schema grouped into rows.
             private readonly string[] _keepColumns;
 
+            // GroupIds[i] is the i-th group-key column's column index in the source schema.
             public readonly int[] GroupIds;
+            // GroupIds[i] is the i-th grouped column's column index in the source schema.
             public readonly int[] KeepIds;
 
             private readonly int _groupCount;
-            private readonly ColumnType[] _columnTypes;
+            private readonly ColumnType[] _keepColumnTypes;
 
             private readonly Dictionary<string, int> _columnNameMap;
 
@@ -224,10 +228,20 @@ public GroupSchema(IExceptionContext ectx, Schema inputSchema, string[] groupCol
                 _keepColumns = keepColumns;
                 KeepIds = GetColumnIds(inputSchema, keepColumns, x => _ectx.ExceptUserArg(nameof(Arguments.Column), x));
 
-                _columnTypes = BuildColumnTypes(_input, KeepIds);
+                _keepColumnTypes = BuildColumnTypes(_input, KeepIds);
                 _columnNameMap = BuildColumnNameMap();
 
-                AsSchema = Schema.Create(this);
+                var schemaBuilder = new SchemaBuilder();
+                foreach (var groupKeyColumnName in groupColumns)
+                    schemaBuilder.AddColumn(groupKeyColumnName, inputSchema[groupKeyColumnName].Type, inputSchema[groupKeyColumnName].Metadata);
+                foreach (var groupValueColumnName in keepColumns)
+                {
+                    var metadataBuilder = new MetadataBuilder();
+                    metadataBuilder.Add(inputSchema[groupValueColumnName].Metadata,
+                        s => s == MetadataUtils.Kinds.IsNormalized || s == MetadataUtils.Kinds.KeyValues);
+                    schemaBuilder.AddColumn(groupValueColumnName, inputSchema[groupValueColumnName].Type, metadataBuilder.GetMetadata());
+                }
+                AsSchema = schemaBuilder.GetSchema();
             }
 
             public GroupSchema(Schema inputSchema, IHostEnvironment env, ModelLoadContext ctx)
@@ -261,7 +275,7 @@ public GroupSchema(Schema inputSchema, IHostEnvironment env, ModelLoadContext ct
 
                 KeepIds = GetColumnIds(inputSchema, _keepColumns, _ectx.Except);
 
-                _columnTypes = BuildColumnTypes(_input, KeepIds);
+                _keepColumnTypes = BuildColumnTypes(_input, KeepIds);
                 _columnNameMap = BuildColumnNameMap();
 
                 AsSchema = Schema.Create(this);
@@ -319,23 +333,34 @@ public void Save(ModelSaveContext ctx)
                 }
             }
 
+            /// <summary>
+            /// Given column names, extract and return column indexes from source schema.
+            /// </summary>
+            /// <param name="schema">Source schema</param>
+            /// <param name="names">Column names</param>
+            /// <param name="except">Marked exception function</param>
+            /// <returns>column indexes</returns>
             private int[] GetColumnIds(Schema schema, string[] names, Func<string, Exception> except)
             {
                 Contracts.AssertValue(schema);
                 Contracts.AssertValue(names);
 
                 var ids = new int[names.Length];
+
                 for (int i = 0; i < names.Length; i++)
                 {
-                    int col;
-                    if (!schema.TryGetColumnIndex(names[i], out col))
+                    // Find column called names[i] from input schema.
+                    var retrievedColumn = schema.GetColumnOrNull(names[i]);
+
+                    // Throw if no such a schema.
+                    if (!retrievedColumn.HasValue)
                         throw except(string.Format("Could not find column '{0}'", names[i]));
 
-                    var colType = schema[col].Type;
+                    var colType = retrievedColumn.Value.Type;
                     if (!colType.IsPrimitive)
                         throw except(string.Format("Column '{0}' has type '{1}', but must have a primitive type", names[i], colType));
 
-                    ids[i] = col;
+                    ids[i] = retrievedColumn.Value.Index;
                 }
 
                 return ids;
@@ -366,7 +391,7 @@ public ColumnType GetColumnType(int col)
                 CheckColumnInRange(col);
                 if (col < _groupCount)
                     return _input[GroupIds[col]].Type;
-                return _columnTypes[col - _groupCount];
+                return _keepColumnTypes[col - _groupCount];
             }
 
             public IEnumerable<KeyValuePair<string, ColumnType>> GetMetadataTypes(int col)