diff --git a/build.gradle b/build.gradle index 2103c55116f9..e132491b1713 100644 --- a/build.gradle +++ b/build.gradle @@ -79,6 +79,7 @@ subprojects { all { exclude group: 'org.slf4j', module: 'slf4j-log4j12' exclude group: 'org.mortbay.jetty' + exclude group: 'org.pentaho', module: 'pentaho-aggdesigner-algorithm' resolutionStrategy { force 'com.fasterxml.jackson.module:jackson-module-scala_2.11:2.10.2' @@ -332,11 +333,34 @@ project(':iceberg-mr') { compileOnly("org.apache.hadoop:hadoop-client") { exclude group: 'org.apache.avro', module: 'avro' } + + compileOnly("org.apache.hive:hive-exec::core") { + exclude group: 'com.google.code.findbugs', module: 'jsr305' + exclude group: 'com.google.guava' + exclude group: 'com.google.protobuf', module: 'protobuf-java' + exclude group: 'org.apache.avro', module: 'avro' + exclude group: 'org.apache.calcite.avatica' + exclude group: 'org.apache.hive', module: 'hive-llap-tez' + exclude group: 'org.apache.logging.log4j' + exclude group: 'org.pentaho' // missing dependency + exclude group: 'org.slf4j', module: 'slf4j-log4j12' + } compileOnly("org.apache.hive:hive-serde") testCompile project(path: ':iceberg-data', configuration: 'testArtifacts') testCompile project(path: ':iceberg-api', configuration: 'testArtifacts') testCompile project(path: ':iceberg-core', configuration: 'testArtifacts') + + testCompile("org.apache.avro:avro:1.9.2") + testCompile("org.apache.calcite:calcite-core") + testCompile("com.esotericsoftware:kryo-shaded:4.0.2") + testCompile("com.fasterxml.jackson.core:jackson-annotations:2.6.5") + testCompile("com.klarna:hiverunner:5.2.1") { + exclude group: 'javax.jms', module: 'jms' + exclude group: 'org.apache.hive', module: 'hive-exec' + exclude group: 'org.codehaus.jettison', module: 'jettison' + exclude group: 'org.apache.calcite.avatica' + } } } diff --git a/mr/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java b/mr/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java index f4b9aae0f55a..e266155522a3 100644 --- a/mr/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java +++ b/mr/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java @@ -19,11 +19,13 @@ package org.apache.iceberg.mr; +import java.io.File; import java.util.function.Function; import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.Schema; import org.apache.iceberg.SchemaParser; import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.expressions.Expression; public class InputFormatConfig { @@ -73,6 +75,10 @@ public ConfigBuilder(Configuration conf) { conf.setBoolean(LOCALITY, false); } + public Configuration conf() { + return conf; + } + public ConfigBuilder filter(Expression expression) { conf.set(FILTER_EXPRESSION, SerializationUtil.serializeToBase64(expression)); return this; @@ -88,6 +94,14 @@ public ConfigBuilder schema(Schema schema) { return this; } + public ConfigBuilder readFrom(TableIdentifier identifier) { + return readFrom(identifier.toString()); + } + + public ConfigBuilder readFrom(File path) { + return readFrom(path.toString()); + } + public ConfigBuilder readFrom(String path) { conf.set(TABLE_PATH, path); return this; diff --git a/mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java b/mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java new file mode 100644 index 000000000000..0b9fd704a08b --- /dev/null +++ b/mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java @@ -0,0 +1,85 @@ +/* + * 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.iceberg.mr.hive; + +import java.io.IOException; +import java.util.Arrays; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; +import org.apache.hadoop.mapred.InputSplit; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.RecordReader; +import org.apache.hadoop.mapred.Reporter; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.Table; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.mr.InputFormatConfig; +import org.apache.iceberg.mr.mapred.Container; +import org.apache.iceberg.mr.mapred.MapredIcebergInputFormat; +import org.apache.iceberg.mr.mapred.TableResolver; +import org.apache.iceberg.mr.mapreduce.IcebergSplit; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +public class HiveIcebergInputFormat extends MapredIcebergInputFormat + implements CombineHiveInputFormat.AvoidSplitCombination { + + private transient Table table; + private transient Schema schema; + + @Override + public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException { + table = TableResolver.resolveTableFromConfiguration(job); + schema = table.schema(); + + forwardConfigSettings(job); + + return Arrays.stream(super.getSplits(job, numSplits)) + .map(split -> new HiveIcebergSplit((IcebergSplit) split, table.location())) + .toArray(InputSplit[]::new); + } + + @Override + public RecordReader> getRecordReader(InputSplit split, JobConf job, + Reporter reporter) throws IOException { + // Since Hive passes a copy of `job` in `getSplits`, we need to forward the conf settings again. + forwardConfigSettings(job); + return super.getRecordReader(split, job, reporter); + } + + @Override + public boolean shouldSkipCombine(Path path, Configuration conf) { + return true; + } + + /** + * Forward configuration settings to the underlying MR input format. + */ + private void forwardConfigSettings(JobConf job) { + Preconditions.checkNotNull(table, "Table cannot be null"); + Preconditions.checkNotNull(schema, "Schema cannot be null"); + + // Once mapred.TableResolver and mapreduce.TableResolver use the same property for the location of the table + // (TABLE_LOCATION vs. TABLE_PATH), this line can be removed: see https://github.com/apache/iceberg/issues/1155. + job.set(InputFormatConfig.TABLE_PATH, table.location()); + job.set(InputFormatConfig.TABLE_SCHEMA, SchemaParser.toJson(schema)); + } +} diff --git a/mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSplit.java b/mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSplit.java new file mode 100644 index 000000000000..e94ff2026d2d --- /dev/null +++ b/mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSplit.java @@ -0,0 +1,93 @@ +/* + * 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.iceberg.mr.hive; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.mapred.FileSplit; +import org.apache.iceberg.mr.SerializationUtil; +import org.apache.iceberg.mr.mapreduce.IcebergSplit; +import org.apache.iceberg.mr.mapreduce.IcebergSplitContainer; + +// Hive requires file formats to return splits that are instances of `FileSplit`. +public class HiveIcebergSplit extends FileSplit implements IcebergSplitContainer { + + private IcebergSplit innerSplit; + + // Hive uses the path name of a split to map it back to a partition (`PartitionDesc`) or table description object + // (`TableDesc`) which specifies the relevant input format for reading the files belonging to that partition or table. + // That way, `HiveInputFormat` and `CombineHiveInputFormat` can read files with different file formats in the same + // MapReduce job and merge compatible splits together. + private String tableLocation; + + // public no-argument constructor for deserialization + public HiveIcebergSplit() {} + + HiveIcebergSplit(IcebergSplit split, String tableLocation) { + this.innerSplit = split; + this.tableLocation = tableLocation; + } + + @Override + public IcebergSplit icebergSplit() { + return innerSplit; + } + + @Override + public long getLength() { + return innerSplit.getLength(); + } + + @Override + public String[] getLocations() { + return innerSplit.getLocations(); + } + + @Override + public Path getPath() { + return new Path(tableLocation); + } + + @Override + public long getStart() { + return 0; + } + + @Override + public void write(DataOutput out) throws IOException { + byte[] bytes = SerializationUtil.serializeToBytes(tableLocation); + out.writeInt(bytes.length); + out.write(bytes); + + innerSplit.write(out); + } + + @Override + public void readFields(DataInput in) throws IOException { + byte[] bytes = new byte[in.readInt()]; + in.readFully(bytes); + tableLocation = SerializationUtil.deserializeFromBytes(bytes); + + innerSplit = new IcebergSplit(); + innerSplit.readFields(in); + } +} diff --git a/mr/src/main/java/org/apache/iceberg/mr/mapred/IcebergWritable.java b/mr/src/main/java/org/apache/iceberg/mr/mapred/Container.java similarity index 64% rename from mr/src/main/java/org/apache/iceberg/mr/mapred/IcebergWritable.java rename to mr/src/main/java/org/apache/iceberg/mr/mapred/Container.java index 1eb67f9d3158..fda174ec9e3a 100644 --- a/mr/src/main/java/org/apache/iceberg/mr/mapred/IcebergWritable.java +++ b/mr/src/main/java/org/apache/iceberg/mr/mapred/Container.java @@ -22,38 +22,22 @@ import java.io.DataInput; import java.io.DataOutput; import org.apache.hadoop.io.Writable; -import org.apache.iceberg.Schema; -import org.apache.iceberg.data.Record; /** - * Wraps an Iceberg Record in a Writable which Hive can use in the SerDe. + * A simple container of objects that you can get and set. + * + * @param the Java type of the object held by this container */ -public class IcebergWritable implements Writable { - - private Record record; - private Schema schema; - - public IcebergWritable(Record record, Schema schema) { - this.record = record; - this.schema = schema; - } +public class Container implements Writable { - @SuppressWarnings("checkstyle:HiddenField") - public void wrapRecord(Record record) { - this.record = record; - } - - public Record record() { - return record; - } + private T value; - public Schema schema() { - return schema; + public T get() { + return value; } - @SuppressWarnings("checkstyle:HiddenField") - public void wrapSchema(Schema schema) { - this.schema = schema; + public void set(T newValue) { + this.value = newValue; } @Override diff --git a/mr/src/main/java/org/apache/iceberg/mr/mapred/IcebergSerDe.java b/mr/src/main/java/org/apache/iceberg/mr/mapred/IcebergSerDe.java index 871e0e677835..f0ee139449fe 100644 --- a/mr/src/main/java/org/apache/iceberg/mr/mapred/IcebergSerDe.java +++ b/mr/src/main/java/org/apache/iceberg/mr/mapred/IcebergSerDe.java @@ -55,7 +55,7 @@ public void initialize(@Nullable Configuration configuration, Properties serDePr @Override public Class getSerializedClass() { - return IcebergWritable.class; + return Container.class; } @Override @@ -70,7 +70,7 @@ public SerDeStats getSerDeStats() { @Override public Object deserialize(Writable writable) { - return ((IcebergWritable) writable).record(); + return ((Container) writable).get(); } @Override diff --git a/mr/src/main/java/org/apache/iceberg/mr/mapred/MapredIcebergInputFormat.java b/mr/src/main/java/org/apache/iceberg/mr/mapred/MapredIcebergInputFormat.java new file mode 100644 index 000000000000..5ecba153336f --- /dev/null +++ b/mr/src/main/java/org/apache/iceberg/mr/mapred/MapredIcebergInputFormat.java @@ -0,0 +1,191 @@ +/* + * 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.iceberg.mr.mapred; + +import java.io.IOException; +import java.util.Optional; +import org.apache.hadoop.mapred.InputFormat; +import org.apache.hadoop.mapred.InputSplit; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.RecordReader; +import org.apache.hadoop.mapred.Reporter; +import org.apache.hadoop.mapreduce.Counter; +import org.apache.hadoop.mapreduce.JobContext; +import org.apache.hadoop.mapreduce.JobID; +import org.apache.hadoop.mapreduce.StatusReporter; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hadoop.mapreduce.TaskAttemptID; +import org.apache.hadoop.mapreduce.task.JobContextImpl; +import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.mr.InputFormatConfig; +import org.apache.iceberg.mr.mapreduce.IcebergSplit; +import org.apache.iceberg.mr.mapreduce.IcebergSplitContainer; + +/** + * Generic MR v1 InputFormat API for Iceberg. + * + * @param Java class of records constructed by Iceberg; default is {@link Record} + */ +public class MapredIcebergInputFormat implements InputFormat> { + + private final org.apache.iceberg.mr.mapreduce.IcebergInputFormat innerInputFormat; + + public MapredIcebergInputFormat() { + this.innerInputFormat = new org.apache.iceberg.mr.mapreduce.IcebergInputFormat<>(); + } + + /** + * Configures the {@code JobConf} to use the {@code MapredIcebergInputFormat} and + * returns a helper to add further configuration. + * + * @param job the {@code JobConf} to configure + */ + public static InputFormatConfig.ConfigBuilder configure(JobConf job) { + job.setInputFormat(MapredIcebergInputFormat.class); + return new InputFormatConfig.ConfigBuilder(job); + } + + @Override + public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException { + return innerInputFormat.getSplits(newJobContext(job)) + .stream() + .map(InputSplit.class::cast) + .toArray(InputSplit[]::new); + } + + @Override + public RecordReader> getRecordReader(InputSplit split, JobConf job, + Reporter reporter) throws IOException { + IcebergSplit icebergSplit = ((IcebergSplitContainer) split).icebergSplit(); + return new MapredIcebergRecordReader<>(innerInputFormat, icebergSplit, job, reporter); + } + + private static final class MapredIcebergRecordReader implements RecordReader> { + + private final org.apache.hadoop.mapreduce.RecordReader innerReader; + private final long splitLength; // for getPos() + + MapredIcebergRecordReader(org.apache.iceberg.mr.mapreduce.IcebergInputFormat mapreduceInputFormat, + IcebergSplit split, JobConf job, Reporter reporter) throws IOException { + TaskAttemptContext context = newTaskAttemptContext(job, reporter); + + try { + innerReader = mapreduceInputFormat.createRecordReader(split, context); + innerReader.initialize(split, context); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + + splitLength = split.getLength(); + } + + @Override + public boolean next(Void key, Container value) throws IOException { + try { + if (innerReader.nextKeyValue()) { + value.set(innerReader.getCurrentValue()); + return true; + } + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new RuntimeException(ie); + } + + return false; + } + + @Override + public Void createKey() { + return null; + } + + @Override + public Container createValue() { + return new Container<>(); + } + + @Override + public long getPos() throws IOException { + return (long) (splitLength * getProgress()); + } + + @Override + public float getProgress() throws IOException { + try { + return innerReader.getProgress(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + + @Override + public void close() throws IOException { + if (innerReader != null) { + innerReader.close(); + } + } + } + + private static JobContext newJobContext(JobConf job) { + JobID jobID = Optional.ofNullable(JobID.forName(job.get(JobContext.ID))) + .orElse(new JobID()); + + return new JobContextImpl(job, jobID); + } + + private static TaskAttemptContext newTaskAttemptContext(JobConf job, Reporter reporter) { + TaskAttemptID taskAttemptID = Optional.ofNullable(TaskAttemptID.forName(job.get(JobContext.TASK_ATTEMPT_ID))) + .orElse(new TaskAttemptID()); + + return new TaskAttemptContextImpl(job, taskAttemptID, toStatusReporter(reporter)); + } + + private static StatusReporter toStatusReporter(Reporter reporter) { + return new StatusReporter() { + @Override + public Counter getCounter(Enum name) { + return reporter.getCounter(name); + } + + @Override + public Counter getCounter(String group, String name) { + return reporter.getCounter(group, name); + } + + @Override + public void progress() { + reporter.progress(); + } + + @Override + public float getProgress() { + return reporter.getProgress(); + } + + @Override + public void setStatus(String status) { + reporter.setStatus(status); + } + }; + } +} diff --git a/mr/src/main/java/org/apache/iceberg/mr/mapred/TableResolver.java b/mr/src/main/java/org/apache/iceberg/mr/mapred/TableResolver.java index 7ee670d5e782..c8c74149e778 100644 --- a/mr/src/main/java/org/apache/iceberg/mr/mapred/TableResolver.java +++ b/mr/src/main/java/org/apache/iceberg/mr/mapred/TableResolver.java @@ -29,7 +29,7 @@ import org.apache.iceberg.mr.InputFormatConfig; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; -final class TableResolver { +public final class TableResolver { private TableResolver() { } @@ -42,7 +42,7 @@ static Table resolveTableFromConfiguration(Configuration conf, Properties proper return resolveTableFromConfiguration(configuration); } - static Table resolveTableFromConfiguration(Configuration conf) throws IOException { + public static Table resolveTableFromConfiguration(Configuration conf) throws IOException { //Default to HadoopTables String catalogName = conf.get(InputFormatConfig.CATALOG_NAME, InputFormatConfig.HADOOP_TABLES); diff --git a/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java index 49c8ddb05bad..8c96c1cb3048 100644 --- a/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java +++ b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java @@ -19,8 +19,6 @@ package org.apache.iceberg.mr.mapreduce; -import java.io.DataInput; -import java.io.DataOutput; import java.io.IOException; import java.io.UncheckedIOException; import java.util.Collections; @@ -31,7 +29,6 @@ import java.util.function.BiFunction; import java.util.function.Function; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.io.Writable; import org.apache.hadoop.mapreduce.InputFormat; import org.apache.hadoop.mapreduce.InputSplit; import org.apache.hadoop.mapreduce.Job; @@ -61,7 +58,6 @@ import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.hadoop.HadoopInputFile; import org.apache.iceberg.hadoop.HadoopTables; -import org.apache.iceberg.hadoop.Util; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.CloseableIterator; import org.apache.iceberg.io.InputFile; @@ -85,8 +81,6 @@ public class IcebergInputFormat extends InputFormat { private static final Logger LOG = LoggerFactory.getLogger(IcebergInputFormat.class); - private transient List splits; - /** * Configures the {@code Job} to use the {@code IcebergInputFormat} and * returns a helper to add further configuration. @@ -100,11 +94,6 @@ public static InputFormatConfig.ConfigBuilder configure(Job job) { @Override public List getSplits(JobContext context) { - if (splits != null) { - LOG.info("Returning cached splits: {}", splits.size()); - return splits; - } - Configuration conf = context.getConfiguration(); Table table = findTable(conf); TableScan scan = table.newScan() @@ -132,7 +121,7 @@ public List getSplits(JobContext context) { scan = scan.filter(filter); } - splits = Lists.newArrayList(); + List splits = Lists.newArrayList(); boolean applyResidual = !conf.getBoolean(InputFormatConfig.SKIP_RESIDUAL_FILTERING, false); InputFormatConfig.InMemoryDataModel model = conf.getEnum(InputFormatConfig.IN_MEMORY_DATA_MODEL, InputFormatConfig.InMemoryDataModel.GENERIC); @@ -184,7 +173,7 @@ private static final class IcebergRecordReader extends RecordReader public void initialize(InputSplit split, TaskAttemptContext newContext) { Configuration conf = newContext.getConfiguration(); // For now IcebergInputFormat does its own split planning and does not accept FileSplit instances - CombinedScanTask task = ((IcebergSplit) split).task; + CombinedScanTask task = ((IcebergSplit) split).task(); this.context = newContext; this.tasks = task.files().iterator(); this.tableSchema = SchemaParser.fromJson(conf.get(InputFormatConfig.TABLE_SCHEMA)); @@ -379,47 +368,4 @@ private static Table findTable(Configuration conf) { } } - static class IcebergSplit extends InputSplit implements Writable { - static final String[] ANYWHERE = new String[]{"*"}; - private CombinedScanTask task; - private transient String[] locations; - private transient Configuration conf; - - IcebergSplit(Configuration conf, CombinedScanTask task) { - this.task = task; - this.conf = conf; - } - - @Override - public long getLength() { - return task.files().stream().mapToLong(FileScanTask::length).sum(); - } - - @Override - public String[] getLocations() { - boolean localityPreferred = conf.getBoolean(InputFormatConfig.LOCALITY, false); - if (!localityPreferred) { - return ANYWHERE; - } - if (locations != null) { - return locations; - } - locations = Util.blockLocations(task, conf); - return locations; - } - - @Override - public void write(DataOutput out) throws IOException { - byte[] data = SerializationUtil.serializeToBytes(this.task); - out.writeInt(data.length); - out.write(data); - } - - @Override - public void readFields(DataInput in) throws IOException { - byte[] data = new byte[in.readInt()]; - in.readFully(data); - this.task = SerializationUtil.deserializeFromBytes(data); - } - } } diff --git a/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergSplit.java b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergSplit.java new file mode 100644 index 000000000000..1e9a6b5938fb --- /dev/null +++ b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergSplit.java @@ -0,0 +1,89 @@ +/* + * 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.iceberg.mr.mapreduce; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.mapreduce.InputSplit; +import org.apache.iceberg.CombinedScanTask; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.hadoop.Util; +import org.apache.iceberg.mr.InputFormatConfig; +import org.apache.iceberg.mr.SerializationUtil; + +// Since this class extends `mapreduce.InputSplit and implements `mapred.InputSplit`, it can be returned by both MR v1 +// and v2 file formats. +public class IcebergSplit extends InputSplit implements org.apache.hadoop.mapred.InputSplit, IcebergSplitContainer { + + public static final String[] ANYWHERE = new String[]{"*"}; + + private CombinedScanTask task; + + private transient String[] locations; + private transient Configuration conf; + + // public no-argument constructor for deserialization + public IcebergSplit() {} + + IcebergSplit(Configuration conf, CombinedScanTask task) { + this.task = task; + this.conf = conf; + } + + public CombinedScanTask task() { + return task; + } + + @Override + public IcebergSplit icebergSplit() { + return this; + } + + @Override + public long getLength() { + return task.files().stream().mapToLong(FileScanTask::length).sum(); + } + + @Override + public String[] getLocations() { + if (locations == null) { + boolean localityPreferred = conf.getBoolean(InputFormatConfig.LOCALITY, false); + locations = localityPreferred ? Util.blockLocations(task, conf) : ANYWHERE; + } + + return locations; + } + + @Override + public void write(DataOutput out) throws IOException { + byte[] data = SerializationUtil.serializeToBytes(this.task); + out.writeInt(data.length); + out.write(data); + } + + @Override + public void readFields(DataInput in) throws IOException { + byte[] data = new byte[in.readInt()]; + in.readFully(data); + this.task = SerializationUtil.deserializeFromBytes(data); + } +} diff --git a/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergSplitContainer.java b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergSplitContainer.java new file mode 100644 index 000000000000..c77543763b1a --- /dev/null +++ b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergSplitContainer.java @@ -0,0 +1,26 @@ +/* + * 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.iceberg.mr.mapreduce; + +public interface IcebergSplitContainer { + + IcebergSplit icebergSplit(); + +} diff --git a/mr/src/test/java/org/apache/iceberg/mr/TestHelper.java b/mr/src/test/java/org/apache/iceberg/mr/TestHelper.java new file mode 100644 index 000000000000..0c968919b9ea --- /dev/null +++ b/mr/src/test/java/org/apache/iceberg/mr/TestHelper.java @@ -0,0 +1,226 @@ +/* + * 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.iceberg.mr; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.Files; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.avro.Avro; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.RandomGenericData; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.data.avro.DataWriter; +import org.apache.iceberg.data.orc.GenericOrcWriter; +import org.apache.iceberg.data.parquet.GenericParquetWriter; +import org.apache.iceberg.hadoop.HadoopTables; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.orc.ORC; +import org.apache.iceberg.parquet.Parquet; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.junit.Assert; +import org.junit.rules.TemporaryFolder; + +public class TestHelper { + + private final Configuration conf; + private final HadoopTables tables; + private final Schema schema; + private final PartitionSpec spec; + private final FileFormat fileFormat; + private final TemporaryFolder tmp; + private final File location; + + private Table table; + + public TestHelper(Configuration conf, HadoopTables tables, Schema schema, PartitionSpec spec, FileFormat fileFormat, + TemporaryFolder tmp, File location) { + this.conf = conf; + this.tables = tables; + this.schema = schema; + this.spec = spec; + this.fileFormat = fileFormat; + this.tmp = tmp; + this.location = location; + } + + private void setTable(Table table) { + this.table = table; + conf.set(InputFormatConfig.TABLE_SCHEMA, SchemaParser.toJson(table.schema())); + } + + public Table getTable() { + return table; + } + + public Table createTable(Schema theSchema, PartitionSpec theSpec) { + Map properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, fileFormat.name()); + Table tbl = tables.create(theSchema, theSpec, properties, location.toString()); + setTable(tbl); + return tbl; + } + + public Table createTable(Catalog catalog, TableIdentifier identifier) { + Map properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, fileFormat.name()); + Table tbl = catalog.createTable(identifier, schema, spec, properties); + setTable(tbl); + return tbl; + } + + public Table createPartitionedTable() { + return createTable(schema, spec); + } + + public Table createUnpartitionedTable() { + return createTable(schema, PartitionSpec.unpartitioned()); + } + + + public List generateRandomRecords(int num, long seed) { + Preconditions.checkNotNull(table, "table not set"); + return generateRandomRecords(table.schema(), num, seed); + } + + public static List generateRandomRecords(Schema schema, int num, long seed) { + return RandomGenericData.generate(schema, num, seed); + } + + public void appendToTable(DataFile... dataFiles) { + Preconditions.checkNotNull(table, "table not set"); + + AppendFiles append = table.newAppend(); + + for (DataFile dataFile : dataFiles) { + append = append.appendFile(dataFile); + } + + append.commit(); + } + + public void appendToTable(StructLike partition, List records) throws IOException { + appendToTable(writeFile(partition, records)); + } + + public DataFile writeFile(StructLike partition, List records) throws IOException { + Preconditions.checkNotNull(table, "table not set"); + return writeFile(table, partition, records, fileFormat, tmp.newFile()); + } + + public static DataFile writeFile(Table table, StructLike partition, List records, FileFormat fileFormat, + File file) throws IOException { + Assert.assertTrue(file.delete()); + + FileAppender appender; + + switch (fileFormat) { + case AVRO: + appender = Avro.write(Files.localOutput(file)) + .schema(table.schema()) + .createWriterFunc(DataWriter::create) + .named(fileFormat.name()) + .build(); + break; + + case PARQUET: + appender = Parquet.write(Files.localOutput(file)) + .schema(table.schema()) + .createWriterFunc(GenericParquetWriter::buildWriter) + .named(fileFormat.name()) + .build(); + break; + + case ORC: + appender = ORC.write(Files.localOutput(file)) + .schema(table.schema()) + .createWriterFunc(GenericOrcWriter::buildWriter) + .build(); + break; + + default: + throw new UnsupportedOperationException("Cannot write format: " + fileFormat); + } + + try { + appender.addAll(records); + } finally { + appender.close(); + } + + DataFiles.Builder builder = DataFiles.builder(table.spec()) + .withPath(file.toString()) + .withFormat(fileFormat) + .withFileSizeInBytes(file.length()) + .withMetrics(appender.metrics()); + + if (partition != null) { + builder.withPartition(partition); + } + + return builder.build(); + } + + public static class RecordsBuilder { + + private final List records = new ArrayList(); + private final Schema schema; + + private RecordsBuilder(Schema schema) { + this.schema = schema; + } + + public RecordsBuilder add(Object... values) { + Preconditions.checkArgument(schema.columns().size() == values.length); + + GenericRecord record = GenericRecord.create(schema); + + for (int i = 0; i < values.length; i++) { + record.set(i, values[i]); + } + + records.add(record); + return this; + } + + public List build() { + return Collections.unmodifiableList(records); + } + + public static RecordsBuilder newInstance(Schema schema) { + return new RecordsBuilder(schema); + } + } +} diff --git a/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java b/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java new file mode 100644 index 000000000000..4470aec0d794 --- /dev/null +++ b/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java @@ -0,0 +1,485 @@ +/* + * 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.iceberg.mr; + +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.Reporter; +import org.apache.hadoop.mapred.TaskAttemptID; +import org.apache.hadoop.mapreduce.InputSplit; +import org.apache.hadoop.mapreduce.RecordReader; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.AssertHelpers; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.TestHelpers.Row; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.hadoop.HadoopTables; +import org.apache.iceberg.mr.mapred.Container; +import org.apache.iceberg.mr.mapred.MapredIcebergInputFormat; +import org.apache.iceberg.mr.mapreduce.IcebergInputFormat; +import org.apache.iceberg.mr.mapreduce.IcebergSplit; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import static org.apache.iceberg.types.Types.NestedField.required; + +@RunWith(Parameterized.class) +public class TestIcebergInputFormats { + + private static final List> TESTED_INPUT_FORMATS = ImmutableList.of( + TestInputFormat.newFactory("IcebergInputFormat", TestIcebergInputFormat::create), + TestInputFormat.newFactory("MapredIcebergInputFormat", TestMapredIcebergInputFormat::create)); + + private static final List TESTED_FILE_FORMATS = ImmutableList.of("avro", "orc", "parquet"); + + private static final Schema SCHEMA = new Schema( + required(1, "data", Types.StringType.get()), + required(2, "id", Types.LongType.get()), + required(3, "date", Types.StringType.get())); + + private static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA) + .identity("date") + .bucket("id", 1) + .build(); + + @Rule + public TemporaryFolder temp = new TemporaryFolder(); + + // before variables + private Configuration conf; + private TestHelper helper; + private InputFormatConfig.ConfigBuilder builder; + + // parametrized variables + private final TestInputFormat.Factory testInputFormat; + private final FileFormat fileFormat; + + @Before + public void before() throws IOException { + conf = new Configuration(); + HadoopTables tables = new HadoopTables(conf); + + File location = temp.newFolder(testInputFormat.name(), fileFormat.name()); + Assert.assertTrue(location.delete()); + + helper = new TestHelper(conf, tables, SCHEMA, SPEC, fileFormat, temp, location); + builder = new InputFormatConfig.ConfigBuilder(conf).readFrom(location); + } + + @Parameterized.Parameters + public static Object[][] parameters() { + Object[][] parameters = new Object[TESTED_INPUT_FORMATS.size() * TESTED_FILE_FORMATS.size()][2]; + + int idx = 0; + + for (TestInputFormat.Factory inputFormat : TESTED_INPUT_FORMATS) { + for (String fileFormat : TESTED_FILE_FORMATS) { + parameters[idx++] = new Object[] {inputFormat, fileFormat}; + } + } + + return parameters; + } + + public TestIcebergInputFormats(TestInputFormat.Factory testInputFormat, String fileFormat) { + this.testInputFormat = testInputFormat; + this.fileFormat = FileFormat.valueOf(fileFormat.toUpperCase(Locale.ENGLISH)); + } + + @Test + public void testUnpartitionedTable() throws Exception { + helper.createUnpartitionedTable(); + List expectedRecords = helper.generateRandomRecords(1, 0L); + helper.appendToTable(null, expectedRecords); + + testInputFormat.create(builder.conf()).validate(expectedRecords); + } + + @Test + public void testPartitionedTable() throws Exception { + helper.createPartitionedTable(); + List expectedRecords = helper.generateRandomRecords(1, 0L); + expectedRecords.get(0).set(2, "2020-03-20"); + helper.appendToTable(Row.of("2020-03-20", 0), expectedRecords); + + testInputFormat.create(builder.conf()).validate(expectedRecords); + } + + @Test + public void testFilterExp() throws Exception { + helper.createPartitionedTable(); + + List expectedRecords = helper.generateRandomRecords(2, 0L); + expectedRecords.get(0).set(2, "2020-03-20"); + expectedRecords.get(1).set(2, "2020-03-20"); + + DataFile dataFile1 = helper.writeFile(Row.of("2020-03-20", 0), expectedRecords); + DataFile dataFile2 = helper.writeFile(Row.of("2020-03-21", 0), helper.generateRandomRecords(2, 0L)); + helper.appendToTable(dataFile1, dataFile2); + + builder.filter(Expressions.equal("date", "2020-03-20")); + testInputFormat.create(builder.conf()).validate(expectedRecords); + } + + @Test + public void testResiduals() throws Exception { + helper.createPartitionedTable(); + + List writeRecords = helper.generateRandomRecords(2, 0L); + writeRecords.get(0).set(1, 123L); + writeRecords.get(0).set(2, "2020-03-20"); + writeRecords.get(1).set(1, 456L); + writeRecords.get(1).set(2, "2020-03-20"); + + List expectedRecords = new ArrayList<>(); + expectedRecords.add(writeRecords.get(0)); + + DataFile dataFile1 = helper.writeFile(Row.of("2020-03-20", 0), writeRecords); + DataFile dataFile2 = helper.writeFile(Row.of("2020-03-21", 0), helper.generateRandomRecords(2, 0L)); + helper.appendToTable(dataFile1, dataFile2); + + builder.filter(Expressions.and( + Expressions.equal("date", "2020-03-20"), + Expressions.equal("id", 123))); + testInputFormat.create(builder.conf()).validate(expectedRecords); + + // skip residual filtering + builder.skipResidualFiltering(); + testInputFormat.create(builder.conf()).validate(writeRecords); + } + + @Test + public void testFailedResidualFiltering() throws Exception { + helper.createPartitionedTable(); + + List expectedRecords = helper.generateRandomRecords(2, 0L); + expectedRecords.get(0).set(2, "2020-03-20"); + expectedRecords.get(1).set(2, "2020-03-20"); + + helper.appendToTable(Row.of("2020-03-20", 0), expectedRecords); + + builder.useHiveRows() + .filter(Expressions.and( + Expressions.equal("date", "2020-03-20"), + Expressions.equal("id", 0))); + + AssertHelpers.assertThrows( + "Residuals are not evaluated today for Iceberg Generics In memory model of HIVE", + UnsupportedOperationException.class, "Filter expression ref(name=\"id\") == 0 is not completely satisfied.", + () -> testInputFormat.create(builder.conf())); + + builder.usePigTuples(); + + AssertHelpers.assertThrows( + "Residuals are not evaluated today for Iceberg Generics In memory model of PIG", + UnsupportedOperationException.class, "Filter expression ref(name=\"id\") == 0 is not completely satisfied.", + () -> testInputFormat.create(builder.conf())); + } + + @Test + public void testProjection() throws Exception { + helper.createPartitionedTable(); + List inputRecords = helper.generateRandomRecords(1, 0L); + helper.appendToTable(Row.of("2020-03-20", 0), inputRecords); + + Schema projection = TypeUtil.select(SCHEMA, ImmutableSet.of(1)); + builder.project(projection); + + List outputRecords = testInputFormat.create(builder.conf()).getRecords(); + + Assert.assertEquals(inputRecords.size(), outputRecords.size()); + Assert.assertEquals(projection.asStruct(), outputRecords.get(0).struct()); + } + + private static final Schema LOG_SCHEMA = new Schema( + Types.NestedField.optional(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "date", Types.StringType.get()), + Types.NestedField.optional(3, "level", Types.StringType.get()), + Types.NestedField.optional(4, "message", Types.StringType.get()) + ); + + private static final PartitionSpec IDENTITY_PARTITION_SPEC = + PartitionSpec.builderFor(LOG_SCHEMA).identity("date").identity("level").build(); + + @Test + public void testIdentityPartitionProjections() throws Exception { + helper.createTable(LOG_SCHEMA, IDENTITY_PARTITION_SPEC); + List inputRecords = helper.generateRandomRecords(10, 0L); + + Integer idx = 0; + AppendFiles append = helper.getTable().newAppend(); + for (Record record : inputRecords) { + record.set(1, "2020-03-2" + idx); + record.set(2, idx.toString()); + append.appendFile(helper.writeFile(Row.of("2020-03-2" + idx, idx.toString()), ImmutableList.of(record))); + idx += 1; + } + append.commit(); + + // individual fields + validateIdentityPartitionProjections(withColumns("date"), inputRecords); + validateIdentityPartitionProjections(withColumns("level"), inputRecords); + validateIdentityPartitionProjections(withColumns("message"), inputRecords); + validateIdentityPartitionProjections(withColumns("id"), inputRecords); + // field pairs + validateIdentityPartitionProjections(withColumns("date", "message"), inputRecords); + validateIdentityPartitionProjections(withColumns("level", "message"), inputRecords); + validateIdentityPartitionProjections(withColumns("date", "level"), inputRecords); + // out-of-order pairs + validateIdentityPartitionProjections(withColumns("message", "date"), inputRecords); + validateIdentityPartitionProjections(withColumns("message", "level"), inputRecords); + validateIdentityPartitionProjections(withColumns("level", "date"), inputRecords); + // full projection + validateIdentityPartitionProjections(LOG_SCHEMA, inputRecords); + // out-of-order triplets + validateIdentityPartitionProjections(withColumns("date", "level", "message"), inputRecords); + validateIdentityPartitionProjections(withColumns("level", "date", "message"), inputRecords); + validateIdentityPartitionProjections(withColumns("date", "message", "level"), inputRecords); + validateIdentityPartitionProjections(withColumns("level", "message", "date"), inputRecords); + validateIdentityPartitionProjections(withColumns("message", "date", "level"), inputRecords); + validateIdentityPartitionProjections(withColumns("message", "level", "date"), inputRecords); + } + + private static Schema withColumns(String... names) { + Map indexByName = TypeUtil.indexByName(LOG_SCHEMA.asStruct()); + Set projectedIds = Sets.newHashSet(); + for (String name : names) { + projectedIds.add(indexByName.get(name)); + } + return TypeUtil.select(LOG_SCHEMA, projectedIds); + } + + private void validateIdentityPartitionProjections(Schema projectedSchema, List inputRecords) { + builder.project(projectedSchema); + List actualRecords = testInputFormat.create(builder.conf()).getRecords(); + + Set fieldNames = TypeUtil.indexByName(projectedSchema.asStruct()).keySet(); + + for (int pos = 0; pos < inputRecords.size(); pos++) { + Record inputRecord = inputRecords.get(pos); + Record actualRecord = actualRecords.get(pos); + Assert.assertEquals("Projected schema should match", projectedSchema.asStruct(), actualRecord.struct()); + + for (String name : fieldNames) { + Assert.assertEquals( + "Projected field " + name + " should match", inputRecord.getField(name), actualRecord.getField(name)); + } + } + } + + @Test + public void testSnapshotReads() throws Exception { + helper.createUnpartitionedTable(); + + List expectedRecords = helper.generateRandomRecords(1, 0L); + helper.appendToTable(null, expectedRecords); + long snapshotId = helper.getTable().currentSnapshot().snapshotId(); + + helper.appendToTable(null, helper.generateRandomRecords(1, 0L)); + + builder.snapshotId(snapshotId); + testInputFormat.create(builder.conf()).validate(expectedRecords); + } + + @Test + public void testLocality() throws Exception { + helper.createUnpartitionedTable(); + List expectedRecords = helper.generateRandomRecords(1, 0L); + helper.appendToTable(null, expectedRecords); + + for (InputSplit split : testInputFormat.create(builder.conf()).getSplits()) { + Assert.assertArrayEquals(IcebergSplit.ANYWHERE, split.getLocations()); + } + + builder.preferLocality(); + + for (InputSplit split : testInputFormat.create(builder.conf()).getSplits()) { + Assert.assertArrayEquals(new String[]{"localhost"}, split.getLocations()); + } + } + + public static class HadoopCatalogLoader implements Function { + @Override + public Catalog apply(Configuration conf) { + return new HadoopCatalog(conf, conf.get("warehouse.location")); + } + } + + @Test + public void testCustomCatalog() throws IOException { + conf.set("warehouse.location", temp.newFolder("hadoop_catalog").getAbsolutePath()); + + Catalog catalog = new HadoopCatalogLoader().apply(conf); + TableIdentifier identifier = TableIdentifier.of("db", "t"); + helper.createTable(catalog, identifier); + + List expectedRecords = helper.generateRandomRecords(1, 0L); + expectedRecords.get(0).set(2, "2020-03-20"); + helper.appendToTable(Row.of("2020-03-20", 0), expectedRecords); + + builder.catalogFunc(HadoopCatalogLoader.class) + .readFrom(identifier); + + testInputFormat.create(builder.conf()).validate(expectedRecords); + } + + private abstract static class TestInputFormat { + + private final List splits; + private final List records; + + private TestInputFormat(List splits, List records) { + this.splits = splits; + this.records = records; + } + + public List getRecords() { + return records; + } + + public List getSplits() { + return splits; + } + + public void validate(List expected) { + Assert.assertEquals(expected, records); + } + + public interface Factory { + String name(); + TestInputFormat create(Configuration conf); + } + + public static Factory newFactory(String name, Function> function) { + return new Factory() { + @Override + public String name() { + return name; + } + + @Override + public TestInputFormat create(Configuration conf) { + return function.apply(conf); + } + }; + } + } + + private static final class TestMapredIcebergInputFormat extends TestInputFormat { + + private TestMapredIcebergInputFormat(List splits, List records) { + super(splits, records); + } + + private static TestMapredIcebergInputFormat create(Configuration conf) { + JobConf job = new JobConf(conf); + MapredIcebergInputFormat inputFormat = new MapredIcebergInputFormat<>(); + + try { + org.apache.hadoop.mapred.InputSplit[] splits = inputFormat.getSplits(job, 1); + + List iceSplits = new ArrayList<>(splits.length); + List records = new ArrayList<>(); + + for (org.apache.hadoop.mapred.InputSplit split : splits) { + iceSplits.add((IcebergSplit) split); + org.apache.hadoop.mapred.RecordReader> + reader = inputFormat.getRecordReader(split, job, Reporter.NULL); + + try { + Container container = reader.createValue(); + + while (reader.next(null, container)) { + records.add(container.get()); + } + } finally { + reader.close(); + } + } + + return new TestMapredIcebergInputFormat<>(iceSplits, records); + } catch (IOException ioe) { + throw new UncheckedIOException(ioe); + } + } + } + + private static final class TestIcebergInputFormat extends TestInputFormat { + + private TestIcebergInputFormat(List splits, List records) { + super(splits, records); + } + + private static TestIcebergInputFormat create(Configuration conf) { + TaskAttemptContext context = new TaskAttemptContextImpl(conf, new TaskAttemptID()); + IcebergInputFormat inputFormat = new IcebergInputFormat<>(); + List splits = inputFormat.getSplits(context); + + List iceSplits = new ArrayList<>(splits.size()); + List records = new ArrayList<>(); + + for (InputSplit split : splits) { + iceSplits.add((IcebergSplit) split); + + try (RecordReader reader = inputFormat.createRecordReader(split, context)) { + reader.initialize(split, context); + + while (reader.nextKeyValue()) { + records.add(reader.getCurrentValue()); + } + } catch (InterruptedException ie) { + throw new RuntimeException(ie); + } catch (IOException ioe) { + throw new UncheckedIOException(ioe); + } + } + + return new TestIcebergInputFormat<>(iceSplits, records); + } + } +} diff --git a/mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergInputFormat.java b/mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergInputFormat.java new file mode 100644 index 000000000000..6443d450453c --- /dev/null +++ b/mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergInputFormat.java @@ -0,0 +1,169 @@ +/* + * 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.iceberg.mr.hive; + +import com.klarna.hiverunner.HiveShell; +import com.klarna.hiverunner.StandaloneHiveRunner; +import com.klarna.hiverunner.annotations.HiveSQL; +import java.io.File; +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.hadoop.HadoopTables; +import org.apache.iceberg.mr.TestHelper; +import org.apache.iceberg.mr.mapred.IcebergSerDe; +import org.apache.iceberg.types.Types; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; + +import static org.apache.iceberg.types.Types.NestedField.required; + +@RunWith(StandaloneHiveRunner.class) +public class TestHiveIcebergInputFormat { + + @HiveSQL(files = {}, autoStart = true) + private HiveShell shell; + + @Rule + public TemporaryFolder temp = new TemporaryFolder(); + + private static final Schema CUSTOMER_SCHEMA = new Schema( + required(1, "customer_id", Types.LongType.get()), + required(2, "first_name", Types.StringType.get()) + ); + + private static final List CUSTOMER_RECORDS = TestHelper.RecordsBuilder.newInstance(CUSTOMER_SCHEMA) + .add(0L, "Alice") + .add(1L, "Bob") + .add(2L, "Trudy") + .build(); + + private static final Schema ORDER_SCHEMA = new Schema( + required(1, "order_id", Types.LongType.get()), + required(2, "customer_id", Types.LongType.get()), + required(3, "total", Types.DoubleType.get())); + + private static final List ORDER_RECORDS = TestHelper.RecordsBuilder.newInstance(ORDER_SCHEMA) + .add(100L, 0L, 11.11d) + .add(101L, 0L, 22.22d) + .add(102L, 1L, 33.33d) + .build(); + + // before variables + private HadoopTables tables; + private Table customerTable; + private Table orderTable; + + @Before + public void before() throws IOException { + Configuration conf = new Configuration(); + tables = new HadoopTables(conf); + + File customerLocation = temp.newFolder("customers"); + Assert.assertTrue(customerLocation.delete()); + + TestHelper customerHelper = new TestHelper( + conf, tables, CUSTOMER_SCHEMA, PartitionSpec.unpartitioned(), FileFormat.PARQUET, temp, customerLocation); + + customerTable = customerHelper.createUnpartitionedTable(); + customerHelper.appendToTable(customerHelper.writeFile(null, CUSTOMER_RECORDS)); + + File orderLocation = temp.newFolder("orders"); + Assert.assertTrue(orderLocation.delete()); + + TestHelper orderHelper = new TestHelper( + conf, tables, ORDER_SCHEMA, PartitionSpec.unpartitioned(), FileFormat.PARQUET, temp, orderLocation); + + orderTable = orderHelper.createUnpartitionedTable(); + orderHelper.appendToTable(orderHelper.writeFile(null, ORDER_RECORDS)); + } + + @Test + public void testScanEmptyTable() throws IOException { + File emptyLocation = temp.newFolder("empty"); + Assert.assertTrue(emptyLocation.delete()); + + Schema emptySchema = new Schema(required(1, "empty", Types.StringType.get())); + Table emptyTable = tables.create( + emptySchema, PartitionSpec.unpartitioned(), Collections.emptyMap(), emptyLocation.toString()); + createHiveTable("empty", emptyTable.location()); + + List rows = shell.executeStatement("SELECT * FROM default.empty"); + Assert.assertEquals(0, rows.size()); + } + + @Test + public void testScanTable() { + createHiveTable("customers", customerTable.location()); + + // Single fetch task: no MR job. + List rows = shell.executeStatement("SELECT * FROM default.customers"); + + Assert.assertEquals(3, rows.size()); + Assert.assertArrayEquals(new Object[] {0L, "Alice"}, rows.get(0)); + Assert.assertArrayEquals(new Object[] {1L, "Bob"}, rows.get(1)); + Assert.assertArrayEquals(new Object[] {2L, "Trudy"}, rows.get(2)); + + // Adding the ORDER BY clause will cause Hive to spawn a local MR job this time. + List descRows = shell.executeStatement("SELECT * FROM default.customers ORDER BY customer_id DESC"); + + Assert.assertEquals(3, rows.size()); + Assert.assertArrayEquals(new Object[] {2L, "Trudy"}, descRows.get(0)); + Assert.assertArrayEquals(new Object[] {1L, "Bob"}, descRows.get(1)); + Assert.assertArrayEquals(new Object[] {0L, "Alice"}, descRows.get(2)); + } + + @Test + public void testJoinTables() { + createHiveTable("customers", customerTable.location()); + createHiveTable("orders", orderTable.location()); + + List rows = shell.executeStatement( + "SELECT c.customer_id, c.first_name, o.order_id, o.total " + + "FROM default.customers c JOIN default.orders o ON c.customer_id = o.customer_id " + + "ORDER BY c.customer_id, o.order_id" + ); + + Assert.assertArrayEquals(new Object[] {0L, "Alice", 100L, 11.11d}, rows.get(0)); + Assert.assertArrayEquals(new Object[] {0L, "Alice", 101L, 22.22d}, rows.get(1)); + Assert.assertArrayEquals(new Object[] {1L, "Bob", 102L, 33.33d}, rows.get(2)); + } + + private void createHiveTable(String table, String location) { + shell.execute(String.format( + "CREATE TABLE default.%s " + + "ROW FORMAT SERDE '%s' " + + "STORED AS " + + "INPUTFORMAT '%s' " + + "OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat' " + + "LOCATION '%s'", + table, IcebergSerDe.class.getName(), HiveIcebergInputFormat.class.getName(), location)); + } +} diff --git a/mr/src/test/java/org/apache/iceberg/mr/mapred/TestHelpers.java b/mr/src/test/java/org/apache/iceberg/mr/mapred/TestHelpers.java deleted file mode 100644 index 0d8cdc036e7f..000000000000 --- a/mr/src/test/java/org/apache/iceberg/mr/mapred/TestHelpers.java +++ /dev/null @@ -1,119 +0,0 @@ -/* - * 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.iceberg.mr.mapred; - -import java.io.File; -import java.io.IOException; -import java.util.List; -import org.apache.iceberg.DataFile; -import org.apache.iceberg.DataFiles; -import org.apache.iceberg.FileFormat; -import org.apache.iceberg.Files; -import org.apache.iceberg.Schema; -import org.apache.iceberg.StructLike; -import org.apache.iceberg.Table; -import org.apache.iceberg.avro.Avro; -import org.apache.iceberg.data.GenericRecord; -import org.apache.iceberg.data.Record; -import org.apache.iceberg.data.avro.DataWriter; -import org.apache.iceberg.data.orc.GenericOrcWriter; -import org.apache.iceberg.data.parquet.GenericParquetWriter; -import org.apache.iceberg.io.FileAppender; -import org.apache.iceberg.orc.ORC; -import org.apache.iceberg.parquet.Parquet; -import org.apache.iceberg.types.Types; - -import static org.apache.iceberg.types.Types.NestedField.optional; -import static org.apache.iceberg.types.Types.NestedField.required; - -public class TestHelpers { - - private TestHelpers() {} - - public static DataFile writeFile(File targetFile, Table table, StructLike partitionData, FileFormat fileFormat, - List records) throws IOException { - if (targetFile.exists()) { - if (!targetFile.delete()) { - throw new IOException("Unable to delete " + targetFile.getAbsolutePath()); - } - } - FileAppender appender; - switch (fileFormat) { - case AVRO: - appender = Avro.write(Files.localOutput(targetFile)) - .schema(table.schema()) - .createWriterFunc(DataWriter::create) - .named(fileFormat.name()) - .build(); - break; - case PARQUET: - appender = Parquet.write(Files.localOutput(targetFile)) - .schema(table.schema()) - .createWriterFunc(GenericParquetWriter::buildWriter) - .named(fileFormat.name()) - .build(); - break; - case ORC: - appender = ORC.write(Files.localOutput(targetFile)) - .schema(table.schema()) - .createWriterFunc(GenericOrcWriter::buildWriter) - .build(); - break; - default: - throw new UnsupportedOperationException("Cannot write format: " + fileFormat); - } - - try { - appender.addAll(records); - } finally { - appender.close(); - } - - DataFiles.Builder builder = DataFiles.builder(table.spec()) - .withPath(targetFile.toString()) - .withFormat(fileFormat) - .withFileSizeInBytes(targetFile.length()) - .withMetrics(appender.metrics()); - if (partitionData != null) { - builder.withPartition(partitionData); - } - return builder.build(); - } - - public static Record createSimpleRecord(long id, String data) { - Schema schema = new Schema(required(1, "id", Types.StringType.get()), - optional(2, "data", Types.LongType.get())); - GenericRecord record = GenericRecord.create(schema); - record.setField("id", id); - record.setField("data", data); - return record; - } - - public static Record createCustomRecord(Schema schema, Object... dataValues) { - GenericRecord record = GenericRecord.create(schema); - List fields = schema.columns(); - for (int i = 0; i < fields.size(); i++) { - record.set(i, dataValues[i]); - } - return record; - } - -} diff --git a/mr/src/test/java/org/apache/iceberg/mr/mapred/TestIcebergSerDe.java b/mr/src/test/java/org/apache/iceberg/mr/mapred/TestIcebergSerDe.java index 937ddba64da2..a2fb90d966fa 100644 --- a/mr/src/test/java/org/apache/iceberg/mr/mapred/TestIcebergSerDe.java +++ b/mr/src/test/java/org/apache/iceberg/mr/mapred/TestIcebergSerDe.java @@ -72,9 +72,10 @@ public void testDeserialize() { IcebergSerDe serDe = new IcebergSerDe(); Record record = RandomGenericData.generate(schema, 1, 0).get(0); - IcebergWritable writable = new IcebergWritable(record, schema); + Container container = new Container<>(); + container.set(record); - Assert.assertEquals(record, serDe.deserialize(writable)); + Assert.assertEquals(record, serDe.deserialize(container)); } } diff --git a/mr/src/test/java/org/apache/iceberg/mr/mapreduce/TestIcebergInputFormat.java b/mr/src/test/java/org/apache/iceberg/mr/mapreduce/TestIcebergInputFormat.java deleted file mode 100644 index 05842f80e6bd..000000000000 --- a/mr/src/test/java/org/apache/iceberg/mr/mapreduce/TestIcebergInputFormat.java +++ /dev/null @@ -1,557 +0,0 @@ -/* - * 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.iceberg.mr.mapreduce; - -import java.io.File; -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; -import java.util.Locale; -import java.util.Map; -import java.util.Set; -import java.util.function.Function; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.mapreduce.InputSplit; -import org.apache.hadoop.mapreduce.Job; -import org.apache.hadoop.mapreduce.RecordReader; -import org.apache.hadoop.mapreduce.TaskAttemptContext; -import org.apache.hadoop.mapreduce.TaskAttemptID; -import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl; -import org.apache.iceberg.AppendFiles; -import org.apache.iceberg.AssertHelpers; -import org.apache.iceberg.DataFile; -import org.apache.iceberg.DataFiles; -import org.apache.iceberg.FileFormat; -import org.apache.iceberg.Files; -import org.apache.iceberg.PartitionSpec; -import org.apache.iceberg.Schema; -import org.apache.iceberg.StructLike; -import org.apache.iceberg.Table; -import org.apache.iceberg.TableProperties; -import org.apache.iceberg.TestHelpers.Row; -import org.apache.iceberg.avro.Avro; -import org.apache.iceberg.catalog.Catalog; -import org.apache.iceberg.catalog.TableIdentifier; -import org.apache.iceberg.data.RandomGenericData; -import org.apache.iceberg.data.Record; -import org.apache.iceberg.data.avro.DataWriter; -import org.apache.iceberg.data.orc.GenericOrcWriter; -import org.apache.iceberg.data.parquet.GenericParquetWriter; -import org.apache.iceberg.expressions.Expressions; -import org.apache.iceberg.hadoop.HadoopCatalog; -import org.apache.iceberg.hadoop.HadoopTables; -import org.apache.iceberg.io.FileAppender; -import org.apache.iceberg.mr.InputFormatConfig; -import org.apache.iceberg.orc.ORC; -import org.apache.iceberg.parquet.Parquet; -import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable; -import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; -import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; -import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; -import org.apache.iceberg.relocated.com.google.common.collect.Sets; -import org.apache.iceberg.types.TypeUtil; -import org.apache.iceberg.types.Types; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; -import org.junit.runner.RunWith; -import org.junit.runners.Parameterized; - -import static org.apache.iceberg.types.Types.NestedField.required; - -@RunWith(Parameterized.class) -public class TestIcebergInputFormat { - static final Schema SCHEMA = new Schema( - required(1, "data", Types.StringType.get()), - required(2, "id", Types.LongType.get()), - required(3, "date", Types.StringType.get())); - - static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA) - .identity("date") - .bucket("id", 1) - .build(); - - @Rule - public TemporaryFolder temp = new TemporaryFolder(); - private HadoopTables tables; - private Configuration conf; - - @Parameterized.Parameters - public static Object[][] parameters() { - return new Object[][]{ - new Object[]{"parquet"}, - new Object[]{"avro"}, - new Object[]{"orc"} - }; - } - - private final FileFormat format; - - public TestIcebergInputFormat(String format) { - this.format = FileFormat.valueOf(format.toUpperCase(Locale.ENGLISH)); - } - - @Before - public void before() { - conf = new Configuration(); - tables = new HadoopTables(conf); - } - - @Test - public void testUnpartitionedTable() throws Exception { - File location = temp.newFolder(format.name()); - Assert.assertTrue(location.delete()); - Table table = tables.create(SCHEMA, PartitionSpec.unpartitioned(), - ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()), - location.toString()); - List expectedRecords = RandomGenericData.generate(table.schema(), 1, 0L); - DataFile dataFile = writeFile(table, null, format, expectedRecords); - table.newAppend() - .appendFile(dataFile) - .commit(); - Job job = Job.getInstance(conf); - InputFormatConfig.ConfigBuilder configBuilder = IcebergInputFormat.configure(job); - configBuilder.readFrom(location.toString()).schema(table.schema()); - validate(job, expectedRecords); - } - - @Test - public void testPartitionedTable() throws Exception { - File location = temp.newFolder(format.name()); - Assert.assertTrue(location.delete()); - Table table = tables.create(SCHEMA, SPEC, - ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()), - location.toString()); - List expectedRecords = RandomGenericData.generate(table.schema(), 1, 0L); - expectedRecords.get(0).set(2, "2020-03-20"); - DataFile dataFile = writeFile(table, Row.of("2020-03-20", 0), format, expectedRecords); - table.newAppend() - .appendFile(dataFile) - .commit(); - - Job job = Job.getInstance(conf); - InputFormatConfig.ConfigBuilder configBuilder = IcebergInputFormat.configure(job); - configBuilder.readFrom(location.toString()).schema(table.schema()); - validate(job, expectedRecords); - } - - @Test - public void testFilterExp() throws Exception { - File location = temp.newFolder(format.name()); - Assert.assertTrue(location.delete()); - Table table = tables.create(SCHEMA, SPEC, - ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()), - location.toString()); - List expectedRecords = RandomGenericData.generate(table.schema(), 2, 0L); - expectedRecords.get(0).set(2, "2020-03-20"); - expectedRecords.get(1).set(2, "2020-03-20"); - DataFile dataFile1 = writeFile(table, Row.of("2020-03-20", 0), format, expectedRecords); - DataFile dataFile2 = writeFile(table, Row.of("2020-03-21", 0), format, - RandomGenericData.generate(table.schema(), 2, 0L)); - table.newAppend() - .appendFile(dataFile1) - .appendFile(dataFile2) - .commit(); - Job job = Job.getInstance(conf); - InputFormatConfig.ConfigBuilder configBuilder = IcebergInputFormat.configure(job); - configBuilder.readFrom(location.toString()) - .schema(table.schema()) - .filter(Expressions.equal("date", "2020-03-20")); - validate(job, expectedRecords); - } - - @Test - public void testResiduals() throws Exception { - File location = temp.newFolder(format.name()); - Assert.assertTrue(location.delete()); - Table table = tables.create(SCHEMA, SPEC, - ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()), - location.toString()); - List writeRecords = RandomGenericData.generate(table.schema(), 2, 0L); - writeRecords.get(0).set(1, 123L); - writeRecords.get(0).set(2, "2020-03-20"); - writeRecords.get(1).set(1, 456L); - writeRecords.get(1).set(2, "2020-03-20"); - - List expectedRecords = new ArrayList<>(); - expectedRecords.add(writeRecords.get(0)); - - DataFile dataFile1 = writeFile(table, Row.of("2020-03-20", 0), format, writeRecords); - DataFile dataFile2 = writeFile(table, Row.of("2020-03-21", 0), format, - RandomGenericData.generate(table.schema(), 2, 0L)); - table.newAppend() - .appendFile(dataFile1) - .appendFile(dataFile2) - .commit(); - Job job = Job.getInstance(conf); - InputFormatConfig.ConfigBuilder configBuilder = IcebergInputFormat.configure(job); - configBuilder.readFrom(location.toString()) - .schema(table.schema()) - .filter(Expressions.and( - Expressions.equal("date", "2020-03-20"), - Expressions.equal("id", 123))); - validate(job, expectedRecords); - - // skip residual filtering - job = Job.getInstance(conf); - configBuilder = IcebergInputFormat.configure(job); - configBuilder.skipResidualFiltering().readFrom(location.toString()) - .schema(table.schema()) - .filter(Expressions.and( - Expressions.equal("date", "2020-03-20"), - Expressions.equal("id", 123))); - validate(job, writeRecords); - } - - @Test - public void testFailedResidualFiltering() throws Exception { - File location = temp.newFolder(format.name()); - Assert.assertTrue(location.delete()); - Table table = tables.create(SCHEMA, SPEC, - ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()), - location.toString()); - List expectedRecords = RandomGenericData.generate(table.schema(), 2, 0L); - expectedRecords.get(0).set(2, "2020-03-20"); - expectedRecords.get(1).set(2, "2020-03-20"); - - DataFile dataFile1 = writeFile(table, Row.of("2020-03-20", 0), format, expectedRecords); - table.newAppend() - .appendFile(dataFile1) - .commit(); - - Job jobShouldFail1 = Job.getInstance(conf); - InputFormatConfig.ConfigBuilder configBuilder = IcebergInputFormat.configure(jobShouldFail1); - configBuilder.useHiveRows().readFrom(location.toString()) - .schema(table.schema()) - .filter(Expressions.and( - Expressions.equal("date", "2020-03-20"), - Expressions.equal("id", 0))); - AssertHelpers.assertThrows( - "Residuals are not evaluated today for Iceberg Generics In memory model of HIVE", - UnsupportedOperationException.class, "Filter expression ref(name=\"id\") == 0 is not completely satisfied.", - () -> validate(jobShouldFail1, expectedRecords)); - - Job jobShouldFail2 = Job.getInstance(conf); - configBuilder = IcebergInputFormat.configure(jobShouldFail2); - configBuilder.usePigTuples().readFrom(location.toString()) - .schema(table.schema()) - .filter(Expressions.and( - Expressions.equal("date", "2020-03-20"), - Expressions.equal("id", 0))); - AssertHelpers.assertThrows( - "Residuals are not evaluated today for Iceberg Generics In memory model of PIG", - UnsupportedOperationException.class, "Filter expression ref(name=\"id\") == 0 is not completely satisfied.", - () -> validate(jobShouldFail2, expectedRecords)); - } - - @Test - public void testProjection() throws Exception { - File location = temp.newFolder(format.name()); - Assert.assertTrue(location.delete()); - Schema projectedSchema = TypeUtil.select(SCHEMA, ImmutableSet.of(1)); - Table table = tables.create(SCHEMA, SPEC, - ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()), - location.toString()); - List inputRecords = RandomGenericData.generate(table.schema(), 1, 0L); - DataFile dataFile = writeFile(table, Row.of("2020-03-20", 0), format, inputRecords); - table.newAppend() - .appendFile(dataFile) - .commit(); - - Job job = Job.getInstance(conf); - InputFormatConfig.ConfigBuilder configBuilder = IcebergInputFormat.configure(job); - configBuilder - .schema(table.schema()) - .readFrom(location.toString()) - .project(projectedSchema); - List outputRecords = readRecords(job.getConfiguration()); - Assert.assertEquals(inputRecords.size(), outputRecords.size()); - Assert.assertEquals(projectedSchema.asStruct(), outputRecords.get(0).struct()); - } - - private static final Schema LOG_SCHEMA = new Schema( - Types.NestedField.optional(1, "id", Types.IntegerType.get()), - Types.NestedField.optional(2, "date", Types.StringType.get()), - Types.NestedField.optional(3, "level", Types.StringType.get()), - Types.NestedField.optional(4, "message", Types.StringType.get()) - ); - - private static final PartitionSpec IDENTITY_PARTITION_SPEC = - PartitionSpec.builderFor(LOG_SCHEMA).identity("date").identity("level").build(); - - @Test - public void testIdentityPartitionProjections() throws Exception { - File location = temp.newFolder(format.name()); - Assert.assertTrue(location.delete()); - Table table = tables.create(LOG_SCHEMA, IDENTITY_PARTITION_SPEC, - ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()), - location.toString()); - - List inputRecords = RandomGenericData.generate(LOG_SCHEMA, 10, 0); - Integer idx = 0; - AppendFiles append = table.newAppend(); - for (Record record : inputRecords) { - record.set(1, "2020-03-2" + idx); - record.set(2, idx.toString()); - append.appendFile(writeFile(table, Row.of("2020-03-2" + idx, idx.toString()), format, ImmutableList.of(record))); - idx += 1; - } - append.commit(); - - // individual fields - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("date"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("level"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("message"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("id"), inputRecords); - // field pairs - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("date", "message"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("level", "message"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("date", "level"), inputRecords); - // out-of-order pairs - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("message", "date"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("message", "level"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("level", "date"), inputRecords); - // full projection - validateIdentityPartitionProjections(location.toString(), table.schema(), LOG_SCHEMA, inputRecords); - // out-of-order triplets - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("date", "level", "message"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("level", "date", "message"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("date", "message", "level"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("level", "message", "date"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("message", "date", "level"), inputRecords); - validateIdentityPartitionProjections(location.toString(), table.schema(), - withColumns("message", "level", "date"), inputRecords); - } - - private static Schema withColumns(String... names) { - Map indexByName = TypeUtil.indexByName(LOG_SCHEMA.asStruct()); - Set projectedIds = Sets.newHashSet(); - for (String name : names) { - projectedIds.add(indexByName.get(name)); - } - return TypeUtil.select(LOG_SCHEMA, projectedIds); - } - - private void validateIdentityPartitionProjections( - String tablePath, Schema tableSchema, Schema projectedSchema, List inputRecords) throws Exception { - Job job = Job.getInstance(conf); - InputFormatConfig.ConfigBuilder configBuilder = IcebergInputFormat.configure(job); - configBuilder - .schema(tableSchema) - .readFrom(tablePath) - .project(projectedSchema); - List actualRecords = readRecords(job.getConfiguration()); - - Set fieldNames = TypeUtil.indexByName(projectedSchema.asStruct()).keySet(); - for (int pos = 0; pos < inputRecords.size(); pos++) { - Record inputRecord = inputRecords.get(pos); - Record actualRecord = actualRecords.get(pos); - Assert.assertEquals("Projected schema should match", projectedSchema.asStruct(), actualRecord.struct()); - for (String name : fieldNames) { - Assert.assertEquals( - "Projected field " + name + " should match", inputRecord.getField(name), actualRecord.getField(name)); - } - } - } - - @Test - public void testSnapshotReads() throws Exception { - File location = temp.newFolder(format.name()); - Assert.assertTrue(location.delete()); - Table table = tables.create(SCHEMA, PartitionSpec.unpartitioned(), - ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()), - location.toString()); - List expectedRecords = RandomGenericData.generate(table.schema(), 1, 0L); - table.newAppend() - .appendFile(writeFile(table, null, format, expectedRecords)) - .commit(); - long snapshotId = table.currentSnapshot().snapshotId(); - table.newAppend() - .appendFile(writeFile(table, null, format, RandomGenericData.generate(table.schema(), 1, 0L))) - .commit(); - - Job job = Job.getInstance(conf); - InputFormatConfig.ConfigBuilder configBuilder = IcebergInputFormat.configure(job); - configBuilder - .schema(table.schema()) - .readFrom(location.toString()) - .snapshotId(snapshotId); - - validate(job, expectedRecords); - } - - @Test - public void testLocality() throws Exception { - File location = temp.newFolder(format.name()); - Assert.assertTrue(location.delete()); - Table table = tables.create(SCHEMA, PartitionSpec.unpartitioned(), - ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()), - location.toString()); - List expectedRecords = RandomGenericData.generate(table.schema(), 1, 0L); - table.newAppend() - .appendFile(writeFile(table, null, format, expectedRecords)) - .commit(); - Job job = Job.getInstance(conf); - InputFormatConfig.ConfigBuilder configBuilder = IcebergInputFormat.configure(job); - configBuilder.readFrom(location.toString()).schema(table.schema()); - - for (InputSplit split : splits(job.getConfiguration())) { - Assert.assertArrayEquals(IcebergInputFormat.IcebergSplit.ANYWHERE, split.getLocations()); - } - - configBuilder.preferLocality(); - for (InputSplit split : splits(job.getConfiguration())) { - Assert.assertArrayEquals(new String[]{"localhost"}, split.getLocations()); - } - } - - public static class HadoopCatalogFunc implements Function { - @Override - public Catalog apply(Configuration conf) { - return new HadoopCatalog(conf, conf.get("warehouse.location")); - } - } - - @Test - public void testCustomCatalog() throws Exception { - conf = new Configuration(); - conf.set("warehouse.location", temp.newFolder("hadoop_catalog").getAbsolutePath()); - - Catalog catalog = new HadoopCatalogFunc().apply(conf); - TableIdentifier tableIdentifier = TableIdentifier.of("db", "t"); - Table table = catalog.createTable(tableIdentifier, SCHEMA, SPEC, - ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name())); - List expectedRecords = RandomGenericData.generate(table.schema(), 1, 0L); - expectedRecords.get(0).set(2, "2020-03-20"); - DataFile dataFile = writeFile(table, Row.of("2020-03-20", 0), format, expectedRecords); - table.newAppend() - .appendFile(dataFile) - .commit(); - - Job job = Job.getInstance(conf); - InputFormatConfig.ConfigBuilder configBuilder = IcebergInputFormat.configure(job); - configBuilder - .catalogFunc(HadoopCatalogFunc.class) - .schema(table.schema()) - .readFrom(tableIdentifier.toString()); - validate(job, expectedRecords); - } - - private static void validate(Job job, List expectedRecords) { - List actualRecords = readRecords(job.getConfiguration()); - Assert.assertEquals(expectedRecords, actualRecords); - } - - private static List splits(Configuration conf) { - TaskAttemptContext context = new TaskAttemptContextImpl(conf, new TaskAttemptID()); - IcebergInputFormat icebergInputFormat = new IcebergInputFormat<>(); - return icebergInputFormat.getSplits(context); - } - - private static List readRecords(Configuration conf) { - TaskAttemptContext context = new TaskAttemptContextImpl(conf, new TaskAttemptID()); - IcebergInputFormat icebergInputFormat = new IcebergInputFormat<>(); - List splits = icebergInputFormat.getSplits(context); - return - FluentIterable - .from(splits) - .transformAndConcat(split -> readRecords(icebergInputFormat, split, context)) - .toList(); - } - - private static Iterable readRecords( - IcebergInputFormat inputFormat, InputSplit split, TaskAttemptContext context) { - RecordReader recordReader = inputFormat.createRecordReader(split, context); - List records = new ArrayList<>(); - try { - recordReader.initialize(split, context); - while (recordReader.nextKeyValue()) { - records.add(recordReader.getCurrentValue()); - } - } catch (Exception e) { - throw new RuntimeException(e); - } - return records; - } - - private DataFile writeFile( - Table table, StructLike partitionData, FileFormat fileFormat, List records) throws IOException { - File file = temp.newFile(); - Assert.assertTrue(file.delete()); - FileAppender appender; - switch (fileFormat) { - case AVRO: - appender = Avro.write(Files.localOutput(file)) - .schema(table.schema()) - .createWriterFunc(DataWriter::create) - .named(fileFormat.name()) - .build(); - break; - case PARQUET: - appender = Parquet.write(Files.localOutput(file)) - .schema(table.schema()) - .createWriterFunc(GenericParquetWriter::buildWriter) - .named(fileFormat.name()) - .build(); - break; - case ORC: - appender = ORC.write(Files.localOutput(file)) - .schema(table.schema()) - .createWriterFunc(GenericOrcWriter::buildWriter) - .build(); - break; - default: - throw new UnsupportedOperationException("Cannot write format: " + fileFormat); - } - - try { - appender.addAll(records); - } finally { - appender.close(); - } - - DataFiles.Builder builder = DataFiles.builder(table.spec()) - .withPath(file.toString()) - .withFormat(format) - .withFileSizeInBytes(file.length()) - .withMetrics(appender.metrics()); - if (partitionData != null) { - builder.withPartition(partitionData); - } - return builder.build(); - } -}