<commit_before>package org.apache.gora.pig;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;

import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.avro.Schema.Type;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericData.Array;
import org.apache.gora.mapreduce.GoraInputFormat;
import org.apache.gora.mapreduce.GoraInputFormatFactory;
import org.apache.gora.mapreduce.GoraOutputFormat;
import org.apache.gora.mapreduce.GoraOutputFormatFactory;
import org.apache.gora.mapreduce.GoraRecordReader;
import org.apache.gora.mapreduce.GoraRecordWriter;
import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.store.DataStore;
import org.apache.gora.store.DataStoreFactory;
import org.apache.gora.util.AvroUtils;
import org.apache.gora.util.GoraException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.pig.Expression;
import org.apache.pig.LoadCaster;
import org.apache.pig.LoadFunc;
import org.apache.pig.LoadMetadata;
import org.apache.pig.ResourceSchema;
import org.apache.pig.ResourceSchema.ResourceFieldSchema;
import org.apache.pig.ResourceStatistics;
import org.apache.pig.StoreFuncInterface;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.util.UDFContext;
import org.apache.pig.impl.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class GoraStorage extends LoadFunc implements StoreFuncInterface, LoadMetadata {

  public static final Logger LOG = LoggerFactory.getLogger(GoraStorage.class);

  /**
   * Key in UDFContext properties that marks config is set (set at backend nodes)  
   */
  private static final String GORA_CONFIG_SET = "gorastorage.config.set" ;
  private static final String GORA_STORE_SCHEMA = "gorastorage.pig.store.schema" ;

  protected Job job;
  protected JobConf localJobConf ; 
  protected String udfcSignature = null ;
  
  protected String keyClassName ;
  protected String persistentClassName ;
  protected Class<?> keyClass;
  protected Class<? extends PersistentBase> persistentClass;
  protected Schema persistentSchema ;
  private   DataStore<?, ? extends PersistentBase> dataStore ;
  protected GoraInputFormat<?,? extends PersistentBase> inputFormat ;
  protected GoraRecordReader<?,? extends PersistentBase> reader ;
  protected PigGoraOutputFormat<?,? extends PersistentBase> outputFormat ;
  protected GoraRecordWriter<?,? extends PersistentBase> writer ;
  protected PigSplit split ;
  protected ResourceSchema readResourceSchema ;
  protected ResourceSchema writeResourceSchema ;
  private   Map<String, ResourceFieldSchemaWithIndex> writeResourceFieldSchemaMap ;
  
  /** Fields to load as Query - same as {@link loadSaveFields} but without 'key' */
  protected String[] loadQueryFields ;
  
  /** Setted to 'true' if location is '*'. All fields will be loaded into a tuple when reading,
   * and all tuple fields will be copied to the persistent instance when saving. */
  protected boolean loadSaveAllFields = false ;

  /**
   * Creates a new GoraStorage with implicit "*" fields to load/save.  
   * @param keyClassName
   * @param persistentClassName
   * @throws IllegalAccessException 
   * @throws InstantiationException 
   */
  public GoraStorage(String keyClassName, String persistentClassName) throws InstantiationException, IllegalAccessException {
      this(keyClassName, persistentClassName, "*") ;
  }

  /**
   * Creates a new GoraStorage and set the keyClass from the key class name.
   * @param keyClassName key class. Full name with package (org.apache....)
   * @param persistentClassName persistent class. Full name with package. 
   * @param fields comma separated fields to load/save | '*' for all.
   *   '*' loads all fields from the persistent class to each tuple.
   *   '*' saves all fields of each tuple to persist (not mandatory all fields of the persistent class).
   * @throws IllegalAccessException 
   * @throws InstantiationException 
   */
  public GoraStorage(String keyClassName, String persistentClassName, String csvFields) throws InstantiationException, IllegalAccessException {
    super();
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage constructor() {}", this);
    
    this.keyClassName = keyClassName ;
    this.persistentClassName = persistentClassName ;
    try {
      this.keyClass = Class.forName(keyClassName);
      Class<?> persistentClazz = Class.forName(persistentClassName);
      this.persistentClass = persistentClazz.asSubclass(PersistentBase.class);
    } catch (ClassNotFoundException e) {
    	LOG.error("Error creating instance of key and/or persistent.", e) ;
    	throw new RuntimeException(e);
    }

    this.persistentSchema = this.persistentClass.newInstance().getSchema() ;

    // Populates this.loadQueryFields
    List<String> declaredConstructorFields = new ArrayList<String>() ;
    
    if (csvFields.contains("*")) {
      // Declared fields "*"
      this.setLoadSaveAllFields(true) ;
      for (Field field : this.persistentSchema.getFields()) {
        declaredConstructorFields.add(field.name()) ;
      }
    } else {
      // CSV fields declared in constructor.
      String[] fieldsInConstructor = csvFields.split("\\s*,\\s*") ; // splits "field, field, field, field"
      declaredConstructorFields.addAll(Arrays.asList(fieldsInConstructor)) ;
    }

    this.setLoadQueryFields(declaredConstructorFields.toArray(new String[0])) ;

  }

  /**
   * Returns the internal DataStore for <code>&lt;keyClass,persistentClass&gt;</code>
   * using configuration set in job (from setLocation()).
   * Creates one datastore at first call.
   * @return DataStore for &lt;keyClass,persistentClass&gt;
   * @throws GoraException on DataStore creation error.
   */
  protected DataStore<?, ? extends PersistentBase> getDataStore() throws GoraException {
    if (this.localJobConf == null) {
    	LOG.error("Error integrating gora-pig and Pig. Calling getDataStore(). setLocation()/setStoreLocation() must be called first!") ;
    	throw new GoraException("Calling getDataStore(). setLocation()/setStoreLocation() must be called first!") ;
    }
    if (this.dataStore == null) {
      this.dataStore = DataStoreFactory.getDataStore(this.keyClass, this.persistentClass, this.localJobConf) ;
    }
    return this.dataStore ;
  }
  
  /**
   * Gets the job, initialized the localJobConf (the actual used to create a datastore) and splits from 'location' the fields to load/save
   *  
   */
  @Override
  public void setLocation(String location, Job job) throws IOException {
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage setLocation() {} {}", location, this);
    this.job = job;
    this.localJobConf = this.initializeLocalJobConfig(job) ;
  }

  /**
   * Returns UDFProperties based on <code>udfcSignature</code>, <code>keyClassName</code> and <code>persistentClassName</code>.
   */
  private Properties getUDFProperties() {
      return UDFContext.getUDFContext().getUDFProperties(this.getClass(), new String[] {this.udfcSignature,this.keyClassName,this.persistentClassName});
  }
  
  private JobConf initializeLocalJobConfig(Job job) {
    Properties udfProps = getUDFProperties();
    Configuration jobConf = job.getConfiguration();
    JobConf localConf = new JobConf(jobConf); // localConf starts as a copy of jobConf
    if (udfProps.containsKey(GORA_CONFIG_SET)) {
      // Already configured (maybe from frontend to backend)
      for (Entry<Object, Object> entry : udfProps.entrySet()) {
        localConf.set((String) entry.getKey(), (String) entry.getValue());
      }
    } else {
      // Not configured. We load to localConf the configuration and put it in udfProps
      Configuration goraConf = new Configuration();
      for (Entry<String, String> entry : goraConf) {
        // JobConf may have some conf overriding ones in hbase-site.xml
        // So only copy hbase config not in job config to UDFContext
        // Also avoids copying core-default.xml and core-site.xml
        // props in hbaseConf to UDFContext which would be redundant.
        if (localConf.get(entry.getKey()) == null) {
          udfProps.setProperty(entry.getKey(), entry.getValue());
          localConf.set(entry.getKey(), entry.getValue());
        }
      }
      udfProps.setProperty(GORA_CONFIG_SET, "true");
    }
    return localConf;
  }
  
  @Override
  @SuppressWarnings({ "rawtypes", "unchecked" })
  public InputFormat getInputFormat() throws IOException {
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage getInputFormat() {}", this);
    this.inputFormat = GoraInputFormatFactory.createInstance(this.keyClass, this.persistentClass);

    Query query = this.getDataStore().newQuery() ;
    if (this.isLoadSaveAllFields() == false) {
      query.setFields(this.getLoadQueryFields()) ;
    }
    GoraInputFormat.setInput(this.job, query, false) ;
    
    inputFormat.setConf(this.job.getConfiguration()) ;
    return this.inputFormat ; 
  }

  @Override
  public LoadCaster getLoadCaster() throws IOException {
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage getLoadCaster()", this);
    return null;
  }

  @Override
  @SuppressWarnings({ "rawtypes" })
  public void prepareToRead(RecordReader reader, PigSplit split)
      throws IOException {
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage prepareToRead {}", this);
    this.reader = (GoraRecordReader<?, ?>) reader;
    this.split = split;
  }

  @Override
  public Tuple getNext() throws IOException {
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage getNext() {}", this);

    try {
      if (!this.reader.nextKeyValue()) return null;
    } catch (Exception e) {
      LOG.error("Error retrieving next key-value.", e) ;
      throw new IOException(e);
    }
<commit_msg>Añadido Utf8StorageConverter() como Caster de GoraStorage.<commit_after>package org.apache.gora.pig;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;

import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.avro.Schema.Type;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericData.Array;
import org.apache.gora.mapreduce.GoraInputFormat;
import org.apache.gora.mapreduce.GoraInputFormatFactory;
import org.apache.gora.mapreduce.GoraOutputFormat;
import org.apache.gora.mapreduce.GoraOutputFormatFactory;
import org.apache.gora.mapreduce.GoraRecordReader;
import org.apache.gora.mapreduce.GoraRecordWriter;
import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.store.DataStore;
import org.apache.gora.store.DataStoreFactory;
import org.apache.gora.util.AvroUtils;
import org.apache.gora.util.GoraException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.pig.Expression;
import org.apache.pig.LoadCaster;
import org.apache.pig.LoadFunc;
import org.apache.pig.LoadMetadata;
import org.apache.pig.ResourceSchema;
import org.apache.pig.ResourceSchema.ResourceFieldSchema;
import org.apache.pig.ResourceStatistics;
import org.apache.pig.StoreFuncInterface;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.builtin.Utf8StorageConverter;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.util.UDFContext;
import org.apache.pig.impl.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class GoraStorage extends LoadFunc implements StoreFuncInterface, LoadMetadata {

  public static final Logger LOG = LoggerFactory.getLogger(GoraStorage.class);

  /**
   * Key in UDFContext properties that marks config is set (set at backend nodes)  
   */
  private static final String GORA_CONFIG_SET = "gorastorage.config.set" ;
  private static final String GORA_STORE_SCHEMA = "gorastorage.pig.store.schema" ;

  protected Job job;
  protected JobConf localJobConf ; 
  protected String udfcSignature = null ;
  
  protected String keyClassName ;
  protected String persistentClassName ;
  protected Class<?> keyClass;
  protected Class<? extends PersistentBase> persistentClass;
  protected Schema persistentSchema ;
  private   DataStore<?, ? extends PersistentBase> dataStore ;
  protected GoraInputFormat<?,? extends PersistentBase> inputFormat ;
  protected GoraRecordReader<?,? extends PersistentBase> reader ;
  protected PigGoraOutputFormat<?,? extends PersistentBase> outputFormat ;
  protected GoraRecordWriter<?,? extends PersistentBase> writer ;
  protected PigSplit split ;
  protected ResourceSchema readResourceSchema ;
  protected ResourceSchema writeResourceSchema ;
  private   Map<String, ResourceFieldSchemaWithIndex> writeResourceFieldSchemaMap ;
  
  /** Fields to load as Query - same as {@link loadSaveFields} but without 'key' */
  protected String[] loadQueryFields ;
  
  /** Setted to 'true' if location is '*'. All fields will be loaded into a tuple when reading,
   * and all tuple fields will be copied to the persistent instance when saving. */
  protected boolean loadSaveAllFields = false ;

  /**
   * Creates a new GoraStorage with implicit "*" fields to load/save.  
   * @param keyClassName
   * @param persistentClassName
   * @throws IllegalAccessException 
   * @throws InstantiationException 
   */
  public GoraStorage(String keyClassName, String persistentClassName) throws InstantiationException, IllegalAccessException {
      this(keyClassName, persistentClassName, "*") ;
  }

  /**
   * Creates a new GoraStorage and set the keyClass from the key class name.
   * @param keyClassName key class. Full name with package (org.apache....)
   * @param persistentClassName persistent class. Full name with package. 
   * @param fields comma separated fields to load/save | '*' for all.
   *   '*' loads all fields from the persistent class to each tuple.
   *   '*' saves all fields of each tuple to persist (not mandatory all fields of the persistent class).
   * @throws IllegalAccessException 
   * @throws InstantiationException 
   */
  public GoraStorage(String keyClassName, String persistentClassName, String csvFields) throws InstantiationException, IllegalAccessException {
    super();
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage constructor() {}", this);
    
    this.keyClassName = keyClassName ;
    this.persistentClassName = persistentClassName ;
    try {
      this.keyClass = Class.forName(keyClassName);
      Class<?> persistentClazz = Class.forName(persistentClassName);
      this.persistentClass = persistentClazz.asSubclass(PersistentBase.class);
    } catch (ClassNotFoundException e) {
    	LOG.error("Error creating instance of key and/or persistent.", e) ;
    	throw new RuntimeException(e);
    }

    this.persistentSchema = this.persistentClass.newInstance().getSchema() ;

    // Populates this.loadQueryFields
    List<String> declaredConstructorFields = new ArrayList<String>() ;
    
    if (csvFields.contains("*")) {
      // Declared fields "*"
      this.setLoadSaveAllFields(true) ;
      for (Field field : this.persistentSchema.getFields()) {
        declaredConstructorFields.add(field.name()) ;
      }
    } else {
      // CSV fields declared in constructor.
      String[] fieldsInConstructor = csvFields.split("\\s*,\\s*") ; // splits "field, field, field, field"
      declaredConstructorFields.addAll(Arrays.asList(fieldsInConstructor)) ;
    }

    this.setLoadQueryFields(declaredConstructorFields.toArray(new String[0])) ;

  }

  /**
   * Returns the internal DataStore for <code>&lt;keyClass,persistentClass&gt;</code>
   * using configuration set in job (from setLocation()).
   * Creates one datastore at first call.
   * @return DataStore for &lt;keyClass,persistentClass&gt;
   * @throws GoraException on DataStore creation error.
   */
  protected DataStore<?, ? extends PersistentBase> getDataStore() throws GoraException {
    if (this.localJobConf == null) {
    	LOG.error("Error integrating gora-pig and Pig. Calling getDataStore(). setLocation()/setStoreLocation() must be called first!") ;
    	throw new GoraException("Calling getDataStore(). setLocation()/setStoreLocation() must be called first!") ;
    }
    if (this.dataStore == null) {
      this.dataStore = DataStoreFactory.getDataStore(this.keyClass, this.persistentClass, this.localJobConf) ;
    }
    return this.dataStore ;
  }
  
  /**
   * Gets the job, initialized the localJobConf (the actual used to create a datastore) and splits from 'location' the fields to load/save
   *  
   */
  @Override
  public void setLocation(String location, Job job) throws IOException {
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage setLocation() {} {}", location, this);
    this.job = job;
    this.localJobConf = this.initializeLocalJobConfig(job) ;
  }

  /**
   * Returns UDFProperties based on <code>udfcSignature</code>, <code>keyClassName</code> and <code>persistentClassName</code>.
   */
  private Properties getUDFProperties() {
      return UDFContext.getUDFContext().getUDFProperties(this.getClass(), new String[] {this.udfcSignature,this.keyClassName,this.persistentClassName});
  }
  
  private JobConf initializeLocalJobConfig(Job job) {
    Properties udfProps = getUDFProperties();
    Configuration jobConf = job.getConfiguration();
    JobConf localConf = new JobConf(jobConf); // localConf starts as a copy of jobConf
    if (udfProps.containsKey(GORA_CONFIG_SET)) {
      // Already configured (maybe from frontend to backend)
      for (Entry<Object, Object> entry : udfProps.entrySet()) {
        localConf.set((String) entry.getKey(), (String) entry.getValue());
      }
    } else {
      // Not configured. We load to localConf the configuration and put it in udfProps
      Configuration goraConf = new Configuration();
      for (Entry<String, String> entry : goraConf) {
        // JobConf may have some conf overriding ones in hbase-site.xml
        // So only copy hbase config not in job config to UDFContext
        // Also avoids copying core-default.xml and core-site.xml
        // props in hbaseConf to UDFContext which would be redundant.
        if (localConf.get(entry.getKey()) == null) {
          udfProps.setProperty(entry.getKey(), entry.getValue());
          localConf.set(entry.getKey(), entry.getValue());
        }
      }
      udfProps.setProperty(GORA_CONFIG_SET, "true");
    }
    return localConf;
  }
  
  @Override
  @SuppressWarnings({ "rawtypes", "unchecked" })
  public InputFormat getInputFormat() throws IOException {
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage getInputFormat() {}", this);
    this.inputFormat = GoraInputFormatFactory.createInstance(this.keyClass, this.persistentClass);

    Query query = this.getDataStore().newQuery() ;
    if (this.isLoadSaveAllFields() == false) {
      query.setFields(this.getLoadQueryFields()) ;
    }
    GoraInputFormat.setInput(this.job, query, false) ;
    
    inputFormat.setConf(this.job.getConfiguration()) ;
    return this.inputFormat ; 
  }

  @Override
  public LoadCaster getLoadCaster() throws IOException {
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage getLoadCaster()", this);
    return new Utf8StorageConverter();
  }

  @Override
  @SuppressWarnings({ "rawtypes" })
  public void prepareToRead(RecordReader reader, PigSplit split)
      throws IOException {
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage prepareToRead {}", this);
    this.reader = (GoraRecordReader<?, ?>) reader;
    this.split = split;
  }

  @Override
  public Tuple getNext() throws IOException {
    LOG.trace("***"+(UDFContext.getUDFContext().isFrontend()?"[FRONTEND]":"[BACKEND]")+" GoraStorage getNext() {}", this);

    try {
      if (!this.reader.nextKeyValue()) return null;
    } catch (Exception e) {
      LOG.error("Error retrieving next key-value.", e) ;
      throw new IOException(e);
    }

{'fraction_non_alphanumeric': 0.1144262743133057, 'fraction_numerical': 0.00040996674714162076, 'mean_word_length': 4.328640776699029, 'pattern_counts': {'":': 12, '<': 49, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '978673', 'n_tokens_mistral': 6439, 'n_tokens_neox': 6379, 'n_words': 1518}