import org.apache.orc.TypeDescription; //导入依赖的package包/类
@Override
public OrcWriter<T> open(Path path) {
    if (writerOptions == null) {
        writerOptions = OrcFile.writerOptions(configuration);
    }
    if (compressionKind != null) {
        writerOptions.compress(compressionKind);
    }
    if (bufferSize != 0) {
        writerOptions.bufferSize(bufferSize);
    }
    // Add the schema to the writer options.
    TypeDescription schema = getTypeDescription();
    writerOptions.setSchema(schema);
    try {
        writer = OrcFile.createWriter(path, writerOptions);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
    vectorizedRowBatch = schema.createRowBatch(batchSize);
    specialCaseSetup();
    return this;
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
/**
 * @return Length to be used for varchar (or string) data type.
 */
public Integer getLength() {
    if (genInfo.getCategory() == TypeDescription.Category.CHAR) {
        return 1;
    } else if (!GeneratorUtil.isSupportsLengthSpecification(genInfo.getCategory())) {
        return null;
    }
    Orc orc = genInfo.getAnnotation(Orc.class);
    if (orc == null) {
        // Check if jpa column annotation is present.
        Column col = genInfo.getAnnotation(Column.class);
        if (col == null || col.length() == 0) {
            return null;
        }
        return col.length();
    }
    return orc.length() == 0 ? null : orc.length();
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
StringBaseTreeWriter(int columnId, TypeDescription schema, StreamFactory writer,
    boolean nullable) throws IOException {
  super(columnId, schema, writer, nullable);
  this.isDirectV2 = isNewWriteFormat(writer);
  stringOutput = writer.createStream(id, OrcProto.Stream.Kind.DICTIONARY_DATA);
  lengthOutput = createIntegerWriter(writer.createStream(id, OrcProto.Stream.Kind.LENGTH),
      false, isDirectV2, writer);
  rowOutput = createIntegerWriter(writer.createStream(id, OrcProto.Stream.Kind.DATA), false,
      isDirectV2, writer);
  recordPosition(rowIndexPosition);
  rowIndexValueCount.add(0L);
  buildIndex = writer.buildIndex();
  directStreamOutput = writer.createStream(id, OrcProto.Stream.Kind.DATA);
  directLengthOutput = createIntegerWriter(writer.createStream(id, OrcProto.Stream.Kind.LENGTH),
      false, isDirectV2, writer);
  Configuration conf = writer.getConfiguration();
  dictionaryKeySizeThreshold = OrcConf.DICTIONARY_KEY_SIZE_THRESHOLD.getDouble(conf);
  strideDictionaryCheck = OrcConf.ROW_INDEX_STRIDE_DICTIONARY_CHECK.getBoolean(conf);
  doneDictionaryCheck = false;
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
/**
 * Creates an OrcRowInputFormat.
 *
 * @param path The path to read ORC files from.
 * @param orcSchema The schema of the ORC files as ORC TypeDescription.
 * @param orcConfig The configuration to read the ORC files with.
 * @param batchSize The number of Row objects to read in a batch.
 */
public OrcRowInputFormat(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize) {
	super(new Path(path));

	// configure OrcRowInputFormat
	this.schema = orcSchema;
	this.rowType = (RowTypeInfo) OrcUtils.schemaToTypeInfo(schema);
	this.conf = orcConfig;
	this.batchSize = batchSize;

	// set default selection mask, i.e., all fields.
	this.selectedFields = new int[this.schema.getChildren().size()];
	for (int i = 0; i < selectedFields.length; i++) {
		this.selectedFields[i] = i;
	}
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
	batchSize = in.readInt();
	org.apache.hadoop.conf.Configuration configuration = new org.apache.hadoop.conf.Configuration();
	configuration.readFields(in);

	if (this.conf == null) {
		this.conf = configuration;
	}
	this.schema = TypeDescription.fromString(in.readUTF());

	this.selectedFields = new int[in.readInt()];
	for (int i = 0; i < selectedFields.length; i++) {
		this.selectedFields[i] = in.readInt();
	}

	this.conjunctPredicates = new ArrayList<>();
	int numPreds = in.readInt();
	for (int i = 0; i < numPreds; i++) {
		conjunctPredicates.add((Predicate) in.readObject());
	}
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
public JsonORCFileWriter(LogFilePath logFilePath, CompressionCodec codec)
        throws IOException {
    Configuration conf = new Configuration();
    Path path = new Path(logFilePath.getLogFilePath());
    schema = schemaProvider.getSchema(logFilePath.getTopic(),
            logFilePath);
    List<TypeDescription> fieldTypes = schema.getChildren();
    converters = new JsonConverter[fieldTypes.size()];
    for (int c = 0; c < converters.length; ++c) {
        converters[c] = VectorColumnFiller.createConverter(fieldTypes
                .get(c));
    }

    writer = OrcFile.createWriter(path, OrcFile.writerOptions(conf)
            .compress(resolveCompression(codec)).setSchema(schema));
    batch = schema.createRowBatch();
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
private TypeDescription determineOrcType( SchemaDescription.Field f ) {
  switch ( f.pentahoValueMetaType ) {
    case ValueMetaInterface.TYPE_NUMBER:
      return TypeDescription.createDouble();
    case ValueMetaInterface.TYPE_INET:
    case ValueMetaInterface.TYPE_STRING:
      return TypeDescription.createString();
    case ValueMetaInterface.TYPE_BOOLEAN:
      return TypeDescription.createBoolean();
    case ValueMetaInterface.TYPE_INTEGER:
      return TypeDescription.createLong();
    case ValueMetaInterface.TYPE_BIGNUMBER:
      return TypeDescription.createDecimal().withPrecision( 20 ).withScale( 10 );
    case ValueMetaInterface.TYPE_SERIALIZABLE:
      return TypeDescription.createBinary();
    case ValueMetaInterface.TYPE_BINARY:
      return TypeDescription.createBinary();
    case ValueMetaInterface.TYPE_DATE:
      return TypeDescription.createDate();
    case ValueMetaInterface.TYPE_TIMESTAMP:
      return TypeDescription.createTimestamp();
    default:
      throw new RuntimeException( "Field: " + f.formatFieldName + "  Undefined type: " + f.pentahoValueMetaType );
  }
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
/**
 * Convert the Orc TypeDescription Object to our SchemaDescription
 *
 * @param typeDescription
 * @return schemaDescription
 */
public SchemaDescription buildSchemaDescription( TypeDescription typeDescription ) {
  SchemaDescription schemaDesc = new SchemaDescription();
  Iterator fieldNameIterator = typeDescription.getFieldNames().iterator();
  for ( TypeDescription subDescription : typeDescription.getChildren() ) {
    //Assume getFieldNames is 1:1 with getChildren
    String fieldName = (String) fieldNameIterator.next();
    int metaType = determineMetaType( subDescription );
    if ( metaType == -1 ) {
      throw new IllegalStateException(
        "Orc Field Name: " + fieldName + " - Could not find pdi field type for " + subDescription.getCategory()
          .getName() );
    }
    schemaDesc.addField( schemaDesc.new Field( fieldName, fieldName + "",
      determineMetaType( subDescription ), true ) );
  }
  return schemaDesc;
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
private int determineMetaType( TypeDescription subDescription ) {
  switch ( subDescription.getCategory().getName() ) {
    case "string":
      return ValueMetaInterface.TYPE_STRING;
    //TODO: Handle inet
    case "bigint":
      return ValueMetaInterface.TYPE_INTEGER;
    case "double":
      return ValueMetaInterface.TYPE_NUMBER;
    case "decimal":
      return ValueMetaInterface.TYPE_BIGNUMBER;
    case "timestamp":
      return ValueMetaInterface.TYPE_TIMESTAMP;
    case "date":
      return ValueMetaInterface.TYPE_DATE;
    case "boolean":
      return ValueMetaInterface.TYPE_BOOLEAN;
    case "binary":
      return ValueMetaInterface.TYPE_BINARY;
  }
  //if none of the cases match return a -1
  return -1;
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
@VisibleForTesting
RowMetaAndData convertFromOrc( RowMetaAndData rowMetaAndData, VectorizedRowBatch batch, int currentBatchRow,
                                      SchemaDescription schemaDescription, TypeDescription typeDescription,
                                      Map<String, Integer> schemaToOrcSubcripts,
                                      SchemaDescription orcSchemaDescription ) {

  int orcColumn;
  for ( SchemaDescription.Field field : schemaDescription ) {
    SchemaDescription.Field orcField = orcSchemaDescription.getField( field.formatFieldName );
    if ( field != null ) {
      ColumnVector columnVector = batch.cols[ schemaToOrcSubcripts.get( field.pentahoFieldName ) ];
      Object orcToPentahoValue = convertFromSourceToTargetDataType( columnVector, currentBatchRow, orcField.pentahoValueMetaType );

      Object convertToSchemaValue = null;
      try {
        convertToSchemaValue = valueMetaConverter.convertFromSourceToTargetDataType( orcField.pentahoValueMetaType, field.pentahoValueMetaType, orcToPentahoValue );
      } catch ( ValueMetaConversionException e ) {
        logger.error( e );
      }
      rowMetaAndData.addValue( field.pentahoFieldName, field.pentahoValueMetaType, convertToSchemaValue );
    }
  }

  return rowMetaAndData;
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
public PentahoOrcRecordWriter( SchemaDescription schemaDescription, TypeDescription schema, String filePath,
                               Configuration conf ) {
  this.schemaDescription = schemaDescription;
  this.schema = schema;

  final AtomicInteger fieldNumber = new AtomicInteger();  //Mutable field count
  schemaDescription.forEach( field -> setOutputMeta( fieldNumber, field ) );
  outputRowMetaAndData = new RowMetaAndData( outputRowMeta, new Object[ fieldNumber.get() ] );

  try {
    writer = OrcFile.createWriter( new Path( filePath ),
      OrcFile.writerOptions( conf )
        .setSchema( schema ) );
    batch = schema.createRowBatch();
  } catch ( IOException e ) {
    logger.error( e );
  }

  //Write the addition metadata for the fields
  new OrcMetaDataWriter( writer ).write( schemaDescription );
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
/**
 * Create the wrapper that holds column statistics for the specified table descriptor.
 *
 * @param td the table descriptor
 */
public AColumnStatistics(final TableDescriptor td) {
  int j = 0;
  stats = new ColumnStatisticsImpl[td.getNumOfColumns()];
  for (final MColumnDescriptor cd : td.getColumnDescriptors()) {
    final TypeDescription d = ORC_TYPE_MAP.get(cd.getColumnType());
    if (d != null) {
      stats[j] = ColumnStatisticsImpl.create(d);
    }
    j++;
  }
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
/**
 * Create a tree writer.
 * 
 * @param columnId the column id of the column to write
 * @param schema the row schema
 * @param streamFactory limited access to the Writer's data.
 * @param nullable can the value be null?
 * @throws IOException
 */
TreeWriter(int columnId, TypeDescription schema, StreamFactory streamFactory, boolean nullable)
    throws IOException {
  this.streamFactory = streamFactory;
  this.isCompressed = streamFactory.isCompressed();
  this.id = columnId;
  if (nullable) {
    isPresentOutStream = streamFactory.createStream(id, OrcProto.Stream.Kind.PRESENT);
    isPresent = new BitFieldWriter(isPresentOutStream, 1);
  } else {
    isPresent = null;
  }
  this.foundNulls = false;
  createBloomFilter = streamFactory.getBloomFilterColumns()[columnId];
  indexStatistics = ColumnStatisticsImpl.create(schema);
  stripeColStatistics = ColumnStatisticsImpl.create(schema);
  fileStatistics = ColumnStatisticsImpl.create(schema);
  childrenWriters = new TreeWriter[0];
  rowIndex = OrcProto.RowIndex.newBuilder();
  rowIndexEntry = OrcProto.RowIndexEntry.newBuilder();
  rowIndexPosition = new RowIndexPositionRecorder(rowIndexEntry);
  stripeStatsBuilders = Lists.newArrayList();
  if (streamFactory.buildIndex()) {
    rowIndexStream = streamFactory.createStream(id, OrcProto.Stream.Kind.ROW_INDEX);
  } else {
    rowIndexStream = null;
  }
  if (createBloomFilter) {
    bloomFilterEntry = OrcProto.BloomFilter.newBuilder();
    bloomFilterIndex = OrcProto.BloomFilterIndex.newBuilder();
    bloomFilterStream = streamFactory.createStream(id, OrcProto.Stream.Kind.BLOOM_FILTER);
    bloomFilter =
        new BloomFilterIO(streamFactory.getRowIndexStride(), streamFactory.getBloomFilterFPP());
  } else {
    bloomFilterEntry = null;
    bloomFilterIndex = null;
    bloomFilterStream = null;
    bloomFilter = null;
  }
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
BooleanTreeWriter(int columnId, TypeDescription schema, StreamFactory writer, boolean nullable)
    throws IOException {
  super(columnId, schema, writer, nullable);
  PositionedOutputStream out = writer.createStream(id, OrcProto.Stream.Kind.DATA);
  this.writer = new BitFieldWriter(out, 1);
  recordPosition(rowIndexPosition);
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
IntegerTreeWriter(int columnId, TypeDescription schema, StreamFactory writer, boolean nullable)
    throws IOException {
  super(columnId, schema, writer, nullable);
  OutStream out = writer.createStream(id, OrcProto.Stream.Kind.DATA);
  this.isDirectV2 = isNewWriteFormat(writer);
  this.writer = createIntegerWriter(out, true, isDirectV2, writer);
  recordPosition(rowIndexPosition);
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
FloatTreeWriter(int columnId, TypeDescription schema, StreamFactory writer, boolean nullable)
    throws IOException {
  super(columnId, schema, writer, nullable);
  this.stream = writer.createStream(id, OrcProto.Stream.Kind.DATA);
  this.utils = new SerializationUtils();
  recordPosition(rowIndexPosition);
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
DoubleTreeWriter(int columnId, TypeDescription schema, StreamFactory writer, boolean nullable)
    throws IOException {
  super(columnId, schema, writer, nullable);
  this.stream = writer.createStream(id, OrcProto.Stream.Kind.DATA);
  this.utils = new SerializationUtils();
  recordPosition(rowIndexPosition);
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
BinaryTreeWriter(int columnId, TypeDescription schema, StreamFactory writer, boolean nullable)
    throws IOException {
  super(columnId, schema, writer, nullable);
  this.stream = writer.createStream(id, OrcProto.Stream.Kind.DATA);
  this.isDirectV2 = isNewWriteFormat(writer);
  this.length = createIntegerWriter(writer.createStream(id, OrcProto.Stream.Kind.LENGTH), false,
      isDirectV2, writer);
  recordPosition(rowIndexPosition);
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
TimestampTreeWriter(int columnId, TypeDescription schema, StreamFactory writer,
    boolean nullable) throws IOException {
  super(columnId, schema, writer, nullable);
  this.isDirectV2 = isNewWriteFormat(writer);
  this.seconds = createIntegerWriter(writer.createStream(id, OrcProto.Stream.Kind.DATA), true,
      isDirectV2, writer);
  this.nanos = createIntegerWriter(writer.createStream(id, OrcProto.Stream.Kind.SECONDARY),
      false, isDirectV2, writer);
  recordPosition(rowIndexPosition);
  // for unit tests to set different time zones
  this.base_timestamp = Timestamp.valueOf(BASE_TIMESTAMP_STRING).getTime() / MILLIS_PER_SECOND;
  writer.useWriterTimeZone(true);
}
 

import org.apache.orc.TypeDescription; //导入依赖的package包/类
DateTreeWriter(int columnId, TypeDescription schema, StreamFactory writer, boolean nullable)
    throws IOException {
  super(columnId, schema, writer, nullable);
  OutStream out = writer.createStream(id, OrcProto.Stream.Kind.DATA);
  this.isDirectV2 = isNewWriteFormat(writer);
  this.writer = createIntegerWriter(out, true, isDirectV2, writer);
  recordPosition(rowIndexPosition);
}