/**
 * 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.
 */

package org.apache.trevni.avro;

import java.io.IOException;
import java.io.Closeable;
import java.io.File;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.Map;
import java.util.HashMap;
import java.util.List;

import org.apache.trevni.ColumnMetaData;
import org.apache.trevni.ColumnFileReader;
import org.apache.trevni.ColumnValues;
import org.apache.trevni.Input;
import org.apache.trevni.InputFile;
import org.apache.trevni.TrevniRuntimeException;

import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.avro.generic.GenericData;

import static org.apache.trevni.avro.AvroColumnator.isSimple;

/** Read files written with {@link AvroColumnWriter}.  A subset of the schema
 * used for writing may be specified when reading.  In this case only columns
 * of the subset schema are read. */
public class AvroColumnReader<D>
  implements Iterator<D>, Iterable<D>, Closeable {

  private ColumnFileReader reader;
  private GenericData model;
  private Schema fileSchema;
  private Schema readSchema;
  
  private ColumnValues[] values;
  private int[] arrayWidths;
  private int column;                          // current index in values

  /** Parameters for reading an Avro column file. */
  public static class Params {
    Input input;
    Schema schema;
    GenericData model = GenericData.get();

    /** Construct reading from a file. */
    public Params(File file) throws IOException {
      this(new InputFile(file));
    }

    /** Construct reading from input. */
    public Params(Input input) { this.input = input; }

    /** Set subset schema to project data down to. */
    public Params setSchema(Schema schema) {
      this.schema = schema;
      return this;
    }

    /** Set data representation. */
    public Params setModel(GenericData model) {
      this.model = model;
      return this;
    }
  }

  /** Construct a reader for a file. */
  public AvroColumnReader(Params params)
    throws IOException {
    this.reader = new ColumnFileReader(params.input);
    this.model = params.model;
    this.fileSchema =
      Schema.parse(reader.getMetaData().getString(AvroColumnWriter.SCHEMA_KEY));
    this.readSchema = params.schema == null ? fileSchema : params.schema;
    initialize();
  }

  /** Return the schema for data in this file. */
  public Schema getFileSchema() { return fileSchema; }

  void initialize() throws IOException {
    // compute a mapping from column name to number for file
    Map<String,Integer> fileColumnNumbers = new HashMap<String,Integer>();
    int i = 0;
    for (ColumnMetaData c : new AvroColumnator(fileSchema).getColumns())
      fileColumnNumbers.put(c.getName(), i++);

    // create iterator for each column in readSchema
    AvroColumnator readColumnator = new AvroColumnator(readSchema);
    this.arrayWidths = readColumnator.getArrayWidths();
    ColumnMetaData[] readColumns = readColumnator.getColumns();
    this.values = new ColumnValues[readColumns.length];
    int j = 0;
    for (ColumnMetaData c : readColumns) {
      Integer n = fileColumnNumbers.get(c.getName());
      if (n == null)
        throw new TrevniRuntimeException("No column named: "+c.getName());
      values[j++] = reader.getValues(n);
    }
  }

  @Override
  public Iterator<D> iterator() { return this; }

  @Override
  public boolean hasNext() {
    return values[0].hasNext();
  }

  @Override
  public D next() {
    try {
      for (int i = 0; i < values.length; i++)
        values[i].startRow();
      this.column = 0;
      return (D)read(readSchema);
    } catch (IOException e) {
      throw new TrevniRuntimeException(e);
    }
  }

  private Object read(Schema s) throws IOException {
    if (isSimple(s))
      return nextValue(s, column++);

    switch (s.getType()) {
    case MAP: 
      throw new TrevniRuntimeException("Unknown schema: "+s);
    case RECORD: 
      Object record = model.newRecord(null, s);
      for (Field f : s.getFields())
        model.setField(record, f.name(), f.pos(), read(f.schema()));
      return record;
    case ARRAY: 
      int length = values[column].nextLength();
      List elements = (List)new GenericData.Array(length, s);
      int startColumn = column;
      for (int i = 0; i < length; i++) {
        this.column = startColumn;
        Object value = nextValue(s, column++);
        if (!isSimple(s.getElementType()))
          value = read(s.getElementType());
        elements.add(value);
      }
      column = startColumn + arrayWidths[startColumn];
      return elements;
    case UNION:
      Object value = null;
      for (Schema branch : s.getTypes()) {
        if (values[column].nextLength() == 1) {
          value = nextValue(s, column);
          column++;
          if (!isSimple(branch))
            value = read(branch);
        } else {
          column += arrayWidths[column];
        }
      }
      return value;
    default:
      throw new TrevniRuntimeException("Unknown schema: "+s);
    }
  }

  private Object nextValue(Schema s, int column) throws IOException {
    Object value = values[column].nextValue();
    
    switch (s.getType()) {
    case ENUM:
      value = GenericData.EnumSymbol(s, s.getEnumSymbols().get((Integer)value));
    case FIXED:
      value = GenericData.Fixed(s, ((ByteBuffer)value).array());
    }

    return value;
  }

  @Override
  public void remove() { throw new UnsupportedOperationException(); }

  @Override
  public void close() throws IOException {
    reader.close();
  }

}

@@ -5862,20 +5862,16 @@
 Object v
-alue
  = value
@@ -5938,39 +5938,42 @@
 ase ENUM:%0A      
-value =
+return new
  GenericData.Enu
@@ -6016,20 +6016,16 @@
 nteger)v
-alue
 ));%0A    
@@ -6042,23 +6042,26 @@
 :%0A      
-value =
+return new
  Generic
@@ -6088,20 +6088,16 @@
 Buffer)v
-alue
 ).array(
@@ -6115,28 +6115,24 @@
     return v
-alue
 ;%0A  %7D%0A%0A  @Ov

Fix a mis-edit.