/* 
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

using ParquetSharp.IO.Api;
using ParquetSharp.Schema;

namespace ParquetSharp.Tools.Read
{
    /**
     * 
     * 
     * @author 
     */
    public class SimpleRecordConverter : GroupConverter
    {
        private static readonly Charset UTF8 = Charset.forName("UTF-8");
        private static readonly CharsetDecoder UTF8_DECODER = UTF8.newDecoder();

        private readonly Converter[] converters;
        private readonly string name;
        private readonly SimpleRecordConverter parent;
        protected SimpleRecord record;

        public SimpleRecordConverter(GroupType schema)
            : this(schema, null, null)
        {
        }

        public SimpleRecordConverter(GroupType schema, string name, SimpleRecordConverter parent)
        {
            this.converters = new Converter[schema.getFieldCount()];
            this.parent = parent;
            this.name = name;

            int i = 0;
            foreach (Type field in schema.getFields())
            {
                converters[i++] = createConverter(field);
            }
        }

        private Converter createConverter(Type field)
        {
            OriginalType? otype = field.getOriginalType();

            if (field.isPrimitive())
            {
                if (otype != null)
                {
                    switch (otype)
                    {
                        case OriginalType.MAP: break;
                        case OriginalType.LIST: break;
                        case OriginalType.UTF8: return new StringConverter(this.record, field.getName());
                        case OriginalType.MAP_KEY_VALUE: break;
                        case OriginalType.ENUM: break;
                    }
                }

                return new SimplePrimitiveConverter(this.record, field.getName());
            }

            GroupType groupType = field.asGroupType();
            if (otype != null)
            {
                switch (otype)
                {
                    case OriginalType.MAP: return new SimpleMapRecordConverter(groupType, field.getName(), this);
                    case OriginalType.LIST: return new SimpleListRecordConverter(groupType, field.getName(), this);
                }
            }
            return new SimpleRecordConverter(groupType, field.getName(), this);
        }

        public override Converter getConverter(int fieldIndex)
        {
            return converters[fieldIndex];
        }

        internal SimpleRecord getCurrentRecord()
        {
            return record;
        }

        public override void start()
        {
            record = new SimpleRecord();
        }

        public override void end()
        {
            if (parent != null)
            {
                parent.getCurrentRecord().add(name, record);
            }
        }

        class SimplePrimitiveConverter : PrimitiveConverter
        {
            protected readonly string name;
            protected readonly SimpleRecord record;

            public SimplePrimitiveConverter(SimpleRecord record, string name)
            {
                this.name = name;
            }

            public override void addBinary(Binary value)
            {
                this.record.add(name, value.getBytes());
            }

            public override void addBoolean(bool value)
            {
                this.record.add(name, value);
            }

            public override void addDouble(double value)
            {
                this.record.add(name, value);
            }

            public override void addFloat(float value)
            {
                this.record.add(name, value);
            }

            public override void addInt(int value)
            {
                this.record.add(name, value);
            }

            public override void addLong(long value)
            {
                this.record.add(name, value);
            }
        }

        class StringConverter : SimplePrimitiveConverter
        {
            public StringConverter(SimpleRecord record, string name)
              : base(record, name)
            {
            }

            public override void addBinary(Binary value)
            {
                this.record.add(name, value.toStringUsingUTF8());
            }
        }
    }
}
