CorcInputFormat implements InputFormat<NullWritable, Corc> { static void setReadColumns(Configuration conf, StructTypeInfo actualStructTypeInfo) { StructTypeInfo readStructTypeInfo = getTypeInfo(conf); LOG.info("Read StructTypeInfo: {}", readStructTypeInfo); List<Integer> ids = new ArrayList<>(); List<String> names = new ArrayList<>(); List<String> readNames = readStructTypeInfo.getAllStructFieldNames(); List<String> actualNames = actualStructTypeInfo.getAllStructFieldNames(); for (int i = 0; i < actualNames.size(); i++) { String actualName = actualNames.get(i); if (readNames.contains(actualName)) { TypeInfo actualTypeInfo = actualStructTypeInfo.getStructFieldTypeInfo(actualName); TypeInfo readTypeInfo = readStructTypeInfo.getStructFieldTypeInfo(actualName); if (!actualTypeInfo.equals(readTypeInfo)) { throw new IllegalStateException("readTypeInfo [" + readTypeInfo + "] does not match actualTypeInfo [" + actualTypeInfo + "]"); } ids.add(i); names.add(actualName); } } if (ids.size() == 0) { throw new IllegalStateException("None of the selected columns were found in the ORC file."); } LOG.info("Set column projection on columns: {} ({})", ids, names); List<String> nestedColumnPaths = Collections.emptyList(); ColumnProjectionUtils.appendReadColumns(conf, ids, names, nestedColumnPaths); }  static void setConverterFactoryClass(Configuration conf,
      Class<? extends ConverterFactory> converterFactoryClass); static void setTypeInfo(Configuration conf, StructTypeInfo typeInfo); static void setSchemaTypeInfo(Configuration conf, StructTypeInfo schemaTypeInfo); static void setSearchArgument(Configuration conf, SearchArgument searchArgument); static void setSearchArgumentKryo(Configuration conf, String searchArgumentKryo); @Override InputSplit[] getSplits(JobConf conf, int numSplits); @Override RecordReader<NullWritable, Corc> getRecordReader(InputSplit inputSplit, JobConf conf, Reporter reporter); static String toKryo(SearchArgument sarg); static final String ENABLE_ROW_LEVEL_SEARCH_ARGUMENT; }
@Test public void setInputReadColumnProjection() { StructTypeInfo typeInfo = new StructTypeInfoBuilder() .add("a", TypeInfoFactory.stringTypeInfo) .add("b", TypeInfoFactory.longTypeInfo) .build(); conf.set(CorcInputFormat.INPUT_TYPE_INFO, "struct<a:string>"); CorcInputFormat.setReadColumns(conf, typeInfo); assertThat(conf.getBoolean(ColumnProjectionUtils.READ_ALL_COLUMNS, true), is(false)); assertThat(conf.get(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR), is("a")); assertThat(conf.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR), is("0")); }
@Test public void setInputReadColumnsAll() { StructTypeInfo typeInfo = new StructTypeInfoBuilder() .add("a", TypeInfoFactory.stringTypeInfo) .add("b", TypeInfoFactory.longTypeInfo) .build(); conf.set(CorcInputFormat.INPUT_TYPE_INFO, "struct<a:string,b:bigint>"); CorcInputFormat.setReadColumns(conf, typeInfo); assertThat(conf.getBoolean(ColumnProjectionUtils.READ_ALL_COLUMNS, true), is(false)); assertThat(conf.get(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR), is("a,b")); assertThat(conf.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR), is("0,1")); }
@Test public void setInputReadColumnsMissing() { StructTypeInfo typeInfo = new StructTypeInfoBuilder() .add("a", TypeInfoFactory.stringTypeInfo) .add("b", TypeInfoFactory.longTypeInfo) .build(); conf.set(CorcInputFormat.INPUT_TYPE_INFO, "struct<a:string,b:bigint,c:string>"); CorcInputFormat.setReadColumns(conf, typeInfo); assertThat(conf.getBoolean(ColumnProjectionUtils.READ_ALL_COLUMNS, true), is(false)); assertThat(conf.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR), is("0,1")); }
@Test(expected = IllegalStateException.class) public void setInputReadColumnsAllMissing() { StructTypeInfo typeInfo = new StructTypeInfoBuilder() .add("a", TypeInfoFactory.stringTypeInfo) .add("b", TypeInfoFactory.longTypeInfo) .build(); conf.set(CorcInputFormat.INPUT_TYPE_INFO, "struct<_col0:string,_col1:bigint>"); CorcInputFormat.setReadColumns(conf, typeInfo); }
@Test(expected = IllegalStateException.class) public void setInputReadColumnsDifferentTypes() { StructTypeInfo typeInfo = new StructTypeInfoBuilder().add("a", TypeInfoFactory.stringTypeInfo).build(); conf.set(CorcInputFormat.INPUT_TYPE_INFO, "struct<a:bigint>"); CorcInputFormat.setReadColumns(conf, typeInfo); }