flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [3/9] flink git commit: [FLINK-2170] [connectors] Add OrcRowInputFormat and OrcTableSource.
Date Wed, 22 Nov 2017 22:10:40 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/edbf8c9e/flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/RowOrcInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/RowOrcInputFormat.java
b/flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/RowOrcInputFormat.java
new file mode 100644
index 0000000..0c9c549
--- /dev/null
+++ b/flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/RowOrcInputFormat.java
@@ -0,0 +1,241 @@
+/*
+ * 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.flink.orc;
+
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.core.fs.FileInputSplit;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.TypeDescription;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+
+import static org.apache.flink.orc.OrcUtils.fillRows;
+
+/**
+ * InputFormat to read ORC data.
+ * For Optimization, reading is done in batch instead of a single row.
+ */
+public class RowOrcInputFormat
+	extends FileInputFormat<Row>
+	implements ResultTypeQueryable<Row> {
+
+	private static final Logger LOG = LoggerFactory.getLogger(RowOrcInputFormat.class);
+	private static final int BATCH_SIZE = 1024;
+
+	private org.apache.hadoop.conf.Configuration config;
+	private TypeDescription schema;
+	private int[] fieldMapping;
+
+	private transient RowTypeInfo rowType;
+	private transient RecordReader orcRowsReader;
+	private transient VectorizedRowBatch rowBatch;
+	private transient Row[] rows;
+
+	private transient int rowInBatch;
+
+	public RowOrcInputFormat(String path, String schemaString, Configuration orcConfig) {
+		this(path, TypeDescription.fromString(schemaString), orcConfig);
+	}
+
+	public RowOrcInputFormat(String path, TypeDescription orcSchema, Configuration orcConfig)
{
+		super(new Path(path));
+		this.unsplittable = false;
+		this.schema = orcSchema;
+		this.rowType = (RowTypeInfo) OrcUtils.schemaToTypeInfo(schema);
+		this.config = orcConfig;
+
+		this.fieldMapping = new int[this.schema.getChildren().size()];
+		for (int i = 0; i < fieldMapping.length; i++) {
+			this.fieldMapping[i] = i;
+		}
+
+	}
+
+	public void setFieldMapping(int[] fieldMapping) {
+		this.fieldMapping = fieldMapping;
+		// adapt result type
+
+		TypeInformation[] fieldTypes = new TypeInformation[fieldMapping.length];
+		String[] fieldNames = new String[fieldMapping.length];
+		for (int i = 0; i < fieldMapping.length; i++) {
+			fieldTypes[i] = this.rowType.getTypeAt(fieldMapping[i]);
+			fieldNames[i] = this.rowType.getFieldNames()[fieldMapping[i]];
+		}
+		this.rowType = new RowTypeInfo(fieldTypes, fieldNames);
+	}
+
+	private boolean[] computeProjectionMask() {
+		boolean[] projectionMask = new boolean[schema.getMaximumId() + 1];
+		for (int inIdx : fieldMapping) {
+			TypeDescription fieldSchema = schema.getChildren().get(inIdx);
+			for (int i = fieldSchema.getId(); i <= fieldSchema.getMaximumId(); i++) {
+				projectionMask[i] = true;
+			}
+		}
+		return projectionMask;
+	}
+
+	@Override
+	public void openInputFormat() throws IOException {
+		super.openInputFormat();
+		this.rows = new Row[BATCH_SIZE];
+		for (int i = 0; i < BATCH_SIZE; i++) {
+			rows[i] = new Row(fieldMapping.length);
+		}
+	}
+
+	@Override
+	public void open(FileInputSplit fileSplit) throws IOException {
+
+		this.currentSplit = fileSplit;
+		Preconditions.checkArgument(this.splitStart == 0, "ORC files must be read from the start.");
+
+		if (LOG.isDebugEnabled()) {
+			LOG.debug("Opening ORC file " + fileSplit.getPath());
+		}
+
+		org.apache.hadoop.fs.Path hPath = new org.apache.hadoop.fs.Path(fileSplit.getPath().getPath());
+
+		Reader orcReader = OrcFile.createReader(hPath, OrcFile.readerOptions(config));
+
+		Reader.Options options = orcReader.options()
+			.range(fileSplit.getStart(), fileSplit.getLength())
+			.useZeroCopy(OrcConf.USE_ZEROCOPY.getBoolean(config))
+			.skipCorruptRecords(OrcConf.SKIP_CORRUPT_DATA.getBoolean(config))
+			.tolerateMissingSchema(OrcConf.TOLERATE_MISSING_SCHEMA.getBoolean(config));
+
+		options.include(computeProjectionMask());
+
+		// check that schema of file is as expected
+		if (!this.schema.equals(orcReader.getSchema())) {
+
+			throw new RuntimeException("Invalid schema for file at " + this.filePath +
+				" Expected:" + this.schema + " Actual: " + orcReader.getSchema());
+		}
+
+		this.orcRowsReader = orcReader.rows(options);
+
+		// assign ids
+		this.schema.getId();
+
+		this.rowBatch = schema.createRowBatch(BATCH_SIZE);
+		rowInBatch = 0;
+	}
+
+	@Override
+	public void close() throws IOException {
+
+		if (orcRowsReader != null) {
+			this.orcRowsReader.close();
+		}
+		this.orcRowsReader = null;
+
+	}
+
+	@Override
+	public void closeInputFormat() throws IOException {
+		this.rows = null;
+		this.rows = null;
+		this.schema = null;
+		this.rowBatch = null;
+	}
+
+	@Override
+	public boolean reachedEnd() throws IOException {
+		return !ensureBatch();
+	}
+
+	private boolean ensureBatch() throws IOException {
+
+		if (rowInBatch >= rowBatch.size) {
+			rowInBatch = 0;
+			boolean moreRows = orcRowsReader.nextBatch(rowBatch);
+
+			if (moreRows) {
+				// read rows
+				fillRows(rows, schema, rowBatch, fieldMapping);
+			}
+			return moreRows;
+		}
+
+		return true;
+	}
+
+	@Override
+	public Row nextRecord(Row reuse) throws IOException {
+		return rows[this.rowInBatch++];
+	}
+
+	@Override
+	public TypeInformation<Row> getProducedType() {
+		return rowType;
+	}
+
+	// --------------------------------------------------------------------------------------------
+	//  Custom serialization methods
+	// --------------------------------------------------------------------------------------------
+
+	private void writeObject(ObjectOutputStream out) throws IOException {
+		this.config.write(out);
+		out.writeUTF(schema.toString());
+
+		out.writeInt(fieldMapping.length);
+		for (int f : fieldMapping) {
+			out.writeInt(f);
+		}
+
+	}
+
+	@SuppressWarnings("unchecked")
+	private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException
{
+
+		org.apache.hadoop.conf.Configuration configuration = new org.apache.hadoop.conf.Configuration();
+		configuration.readFields(in);
+
+		if (this.config == null) {
+			this.config = configuration;
+		}
+		this.schema = TypeDescription.fromString(in.readUTF());
+
+		this.fieldMapping = new int[in.readInt()];
+		for (int i = 0; i < fieldMapping.length; i++) {
+			this.fieldMapping[i] = in.readInt();
+		}
+
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/edbf8c9e/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceITCase.java
b/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceITCase.java
new file mode 100644
index 0000000..3de6ab3
--- /dev/null
+++ b/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceITCase.java
@@ -0,0 +1,142 @@
+/*
+ * 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.flink.orc;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.typeutils.MapTypeInfo;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.java.BatchTableEnvironment;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.apache.flink.types.Row;
+
+import org.hamcrest.CoreMatchers;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Tests for {@link OrcTableSource}.
+ */
+public class OrcTableSourceITCase extends MultipleProgramsTestBase {
+
+	private static final String TEST1_SCHEMA = "struct<boolean1:boolean,byte1:tinyint,short1:smallint,int1:int,"
+
+		"long1:bigint,float1:float,double1:double,bytes1:binary,string1:string," +
+		"middle:struct<list:array<struct<int1:int,string1:string>>>," +
+		"list:array<struct<int1:int,string1:string>>," +
+		"map:map<string,struct<int1:int,string1:string>>>";
+
+	private final URL test1URL = getClass().getClassLoader().getResource("TestOrcFile.test1.orc");
+
+
+	private static final String[] TEST1_DATA = new String[] {
+		"false,1,1024,65536,9223372036854775807,1.0,-15.0,[0, 1, 2, 3, 4],hi,[1,bye, 2,sigh],[3,good,
4,bad],{}",
+		"true,100,2048,65536,9223372036854775807,2.0,-5.0,[],bye,[1,bye, 2,sigh]," +
+			"[100000000,cat, -100000,in, 1234,hat],{chani=5,chani, mauddib=1,mauddib}" };
+
+	public OrcTableSourceITCase() {
+		super(TestExecutionMode.COLLECTION);
+	}
+
+	@Test
+	public void testOrcTableSource() throws Exception {
+
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);
+
+		assert (test1URL != null);
+		OrcTableSource orc = new OrcTableSource(test1URL.getPath(), TEST1_SCHEMA);
+
+		tEnv.registerTableSource("orcTable", orc);
+
+		String query = "Select * from orcTable";
+		Table t = tEnv.sql(query);
+
+		DataSet<Row> dataSet = tEnv.toDataSet(t, Row.class);
+		List<Row> records = dataSet.collect();
+
+		Assert.assertEquals(records.size(), 2);
+
+		List<String> actualRecords = new ArrayList<>();
+		for (Row record : records) {
+			Assert.assertEquals(record.getArity(), 12);
+			actualRecords.add(record.toString());
+		}
+
+		Assert.assertThat(actualRecords, CoreMatchers.hasItems(TEST1_DATA));
+	}
+
+	@Test
+	public void testOrcTableProjection() throws Exception {
+
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);
+
+		assert(test1URL != null);
+		OrcTableSource orc = new OrcTableSource(test1URL.getPath(), TEST1_SCHEMA);
+
+		tEnv.registerTableSource("orcTable", orc);
+
+		String query = "Select middle,list,map from orcTable";
+		Table t = tEnv.sql(query);
+
+		String[] colNames = new String[] {"middle", "list", "map"};
+
+		RowTypeInfo rowTypeInfo = new RowTypeInfo(
+			new TypeInformation[] {
+				BasicTypeInfo.INT_TYPE_INFO,
+				BasicTypeInfo.STRING_TYPE_INFO},
+			new String[] {"int1", "string1"});
+
+		RowTypeInfo structTypeInfo = new RowTypeInfo(
+			new TypeInformation[] {ObjectArrayTypeInfo.getInfoFor(rowTypeInfo)},
+			new String[] {"list"});
+
+		TypeInformation[] colTypes = new TypeInformation[] {
+			structTypeInfo,
+			ObjectArrayTypeInfo.getInfoFor(rowTypeInfo),
+			new MapTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, rowTypeInfo)
+		};
+
+		TableSchema actualTableSchema = new TableSchema(colNames, colTypes);
+
+		Assert.assertArrayEquals(t.getSchema().getColumnNames(), colNames);
+		Assert.assertArrayEquals(t.getSchema().getTypes(), colTypes);
+		Assert.assertEquals(actualTableSchema.toString(), t.getSchema().toString());
+
+		DataSet<Row> dataSet = tEnv.toDataSet(t, Row.class);
+		List<Row> records = dataSet.collect();
+
+		Assert.assertEquals(records.size(), 2);
+		for (Row record: records) {
+			Assert.assertEquals(record.getArity(), 3);
+		}
+
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/edbf8c9e/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java
b/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java
new file mode 100644
index 0000000..c285054
--- /dev/null
+++ b/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.flink.orc;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.typeutils.MapTypeInfo;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.java.BatchTableEnvironment;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.net.URL;
+
+/**
+ * Unit Tests for {@link OrcTableSource}.
+ */
+public class OrcTableSourceTest {
+
+	private static final String TEST1_SCHEMA = "struct<boolean1:boolean,byte1:tinyint,short1:smallint,int1:int,"
+
+		"long1:bigint,float1:float,double1:double,bytes1:binary,string1:string," +
+		"middle:struct<list:array<struct<int1:int,string1:string>>>," +
+		"list:array<struct<int1:int,string1:string>>," +
+		"map:map<string,struct<int1:int,string1:string>>>";
+
+	private final URL test1URL = getClass().getClassLoader().getResource("TestOrcFile.test1.orc");
+
+	@Test
+	public void testOrcSchema() throws Exception {
+
+		assert(test1URL != null);
+		OrcTableSource orc = new OrcTableSource(test1URL.getPath(), TEST1_SCHEMA);
+
+		String expectedSchema = "Row(boolean1: Boolean, byte1: Byte, short1: Short, int1: Integer,
long1: Long, " +
+			"float1: Float, double1: Double, bytes1: byte[], string1: String, " +
+			"middle: Row(list: ObjectArrayTypeInfo<Row(int1: Integer, string1: String)>), "
+
+			"list: ObjectArrayTypeInfo<Row(int1: Integer, string1: String)>, " +
+			"map: Map<String, Row(int1: Integer, string1: String)>)";
+
+		Assert.assertEquals(expectedSchema, orc.getReturnType().toString());
+
+	}
+
+	@Test
+	public void testOrcTableSchema() throws Exception {
+
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);
+
+		assert(test1URL != null);
+		OrcTableSource orc = new OrcTableSource(test1URL.getPath(), TEST1_SCHEMA);
+
+		tEnv.registerTableSource("orcTable", orc);
+		String query = "Select * from orcTable";
+		Table t = tEnv.sql(query);
+
+		String[] colNames = new String[] {
+			"boolean1", "byte1", "short1", "int1", "long1", "float1",
+			"double1", "bytes1", "string1", "list", "list0", "map"
+		};
+
+		RowTypeInfo rowTypeInfo = new RowTypeInfo(
+			new TypeInformation[] {
+				BasicTypeInfo.INT_TYPE_INFO,
+				BasicTypeInfo.STRING_TYPE_INFO},
+			new String[] {"int1", "string1"});
+
+		TypeInformation[] colTypes = new TypeInformation[] {
+			BasicTypeInfo.BOOLEAN_TYPE_INFO,
+			BasicTypeInfo.BYTE_TYPE_INFO,
+			BasicTypeInfo.SHORT_TYPE_INFO,
+			BasicTypeInfo.INT_TYPE_INFO,
+			BasicTypeInfo.LONG_TYPE_INFO,
+			BasicTypeInfo.FLOAT_TYPE_INFO,
+			BasicTypeInfo.DOUBLE_TYPE_INFO,
+			PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO,
+			BasicTypeInfo.STRING_TYPE_INFO,
+			ObjectArrayTypeInfo.getInfoFor(rowTypeInfo),
+			ObjectArrayTypeInfo.getInfoFor(rowTypeInfo),
+			new MapTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, rowTypeInfo)
+		};
+		TableSchema expectedTableSchema = new TableSchema(colNames, colTypes);
+
+		Assert.assertArrayEquals(t.getSchema().getColumnNames(), colNames);
+		Assert.assertArrayEquals(t.getSchema().getTypes(), colTypes);
+		Assert.assertEquals(expectedTableSchema.toString(), t.getSchema().toString());
+
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/edbf8c9e/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/RowOrcInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/RowOrcInputFormatTest.java
b/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/RowOrcInputFormatTest.java
new file mode 100644
index 0000000..60008a0
--- /dev/null
+++ b/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/RowOrcInputFormatTest.java
@@ -0,0 +1,472 @@
+/*
+ * 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.flink.orc;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.core.fs.FileInputSplit;
+import org.apache.flink.types.Row;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Tests for the {@link RowOrcInputFormat}.
+ */
+
+public class RowOrcInputFormatTest {
+
+	private RowOrcInputFormat rowOrcInputFormat;
+
+	@After
+	public void tearDown() throws IOException {
+		if (rowOrcInputFormat != null) {
+			rowOrcInputFormat.close();
+			rowOrcInputFormat.closeInputFormat();
+		}
+		rowOrcInputFormat = null;
+	}
+
+	private final URL test1URL = getClass().getClassLoader().getResource("TestOrcFile.test1.orc");
+
+	private static final String TEST1_SCHEMA = "struct<boolean1:boolean,byte1:tinyint,short1:smallint,int1:int,"
+
+		"long1:bigint,float1:float,double1:double,bytes1:binary,string1:string," +
+		"middle:struct<list:array<struct<int1:int,string1:string>>>," +
+		"list:array<struct<int1:int,string1:string>>," +
+		"map:map<string,struct<int1:int,string1:string>>>";
+
+	private static final String[] TEST1_DATA = new String[] {
+		"false,1,1024,65536,9223372036854775807,1.0,-15.0,[0, 1, 2, 3, 4],hi,[1,bye, 2,sigh],[3,good,
4,bad],{}",
+		"true,100,2048,65536,9223372036854775807,2.0,-5.0,[],bye,[1,bye, 2,sigh]," +
+			"[100000000,cat, -100000,in, 1234,hat],{chani=5,chani, mauddib=1,mauddib}" };
+
+	private static final String[] TEST1_PROJECTED_DATA = new String[] {
+		"{},[3,good, 4,bad],[1,bye, 2,sigh],hi,[0, 1, 2, 3, 4],-15.0,1.0,9223372036854775807,65536,1024,1,false",
+		"{chani=5,chani, mauddib=1,mauddib},[100000000,cat, -100000,in, 1234,hat],[1,bye, 2,sigh],bye,"
+
+			"[],-5.0,2.0,9223372036854775807,65536,2048,100,true" };
+
+	private static final String TEST1_INVALID_SCHEMA = "struct<boolean1:int,byte1:tinyint,short1:smallint,int1:int,"
+
+		"long1:bigint,float1:float,double1:double,bytes1:binary,string1:string," +
+		"middle:struct<list:array<struct<int1:int,string1:string>>>," +
+		"list:array<struct<int1:int,string1:string>>," +
+		"map:map<string,struct<int1:int,string1:string>>>";
+
+	@Test(expected = FileNotFoundException.class)
+	public void testInvalidPath() throws IOException{
+
+		rowOrcInputFormat = new RowOrcInputFormat("TestOrcFile.test2.orc", TEST1_SCHEMA, new Configuration());
+		rowOrcInputFormat.openInputFormat();
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+		rowOrcInputFormat.open(inputSplits[0]);
+
+	}
+
+	@Test(expected = RuntimeException.class)
+	public void testInvalidSchema() throws IOException{
+
+		assert(test1URL != null);
+		rowOrcInputFormat = new RowOrcInputFormat(test1URL.getPath(), TEST1_INVALID_SCHEMA, new
Configuration());
+		rowOrcInputFormat.openInputFormat();
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+		rowOrcInputFormat.open(inputSplits[0]);
+
+	}
+
+	@Test(expected = IndexOutOfBoundsException.class)
+	public void testInvalidProjection() throws IOException{
+
+		assert(test1URL != null);
+		rowOrcInputFormat = new RowOrcInputFormat(test1URL.getPath(), TEST1_SCHEMA, new Configuration());
+		int[] projectionMask = {14};
+		rowOrcInputFormat.setFieldMapping(projectionMask);
+	}
+
+	@Test
+	public void testMajorDataTypes() throws IOException{
+
+		// test for boolean,byte,short,int,long,float,double,bytes,string,struct,list,map
+		assert(test1URL != null);
+		rowOrcInputFormat = new RowOrcInputFormat(test1URL.getPath(), TEST1_SCHEMA, new Configuration());
+		rowOrcInputFormat.openInputFormat();
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		Row row = null;
+		int count = 0;
+		for (FileInputSplit split : inputSplits) {
+			rowOrcInputFormat.open(split);
+			while (!rowOrcInputFormat.reachedEnd()) {
+				row = rowOrcInputFormat.nextRecord(row);
+				Assert.assertEquals(row.toString(), TEST1_DATA[count++]);
+			}
+		}
+	}
+
+	@Test
+	public void testProjection() throws IOException{
+
+		assert(test1URL != null);
+		rowOrcInputFormat = new RowOrcInputFormat(test1URL.getPath(), TEST1_SCHEMA, new Configuration());
+		int[] projectionMask = {11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0};
+		rowOrcInputFormat.setFieldMapping(projectionMask);
+		rowOrcInputFormat.openInputFormat();
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		Row row = null;
+		int count = 0;
+		for (FileInputSplit split : inputSplits) {
+			rowOrcInputFormat.open(split);
+			while (!rowOrcInputFormat.reachedEnd()) {
+				row = rowOrcInputFormat.nextRecord(row);
+				Assert.assertEquals(row.toString(), TEST1_PROJECTED_DATA[count++]);
+			}
+		}
+
+	}
+
+	@Test
+	public void testTimeStampAndDate() throws IOException{
+
+		URL expectedDataURL = getClass().getClassLoader().getResource("TestOrcFile.testDate1900.dat");
+		assert(expectedDataURL != null);
+		List<String> expectedTimeStampAndDate = Files.readAllLines(Paths.get(expectedDataURL.getPath()));
+
+		URL testInputURL = getClass().getClassLoader().getResource("TestOrcFile.testDate1900.orc");
+		assert(testInputURL != null);
+		String path = testInputURL.getPath();
+		String schema = "struct<time:timestamp,date:date>";
+		rowOrcInputFormat = new RowOrcInputFormat(path, schema, new Configuration());
+		rowOrcInputFormat.openInputFormat();
+
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		List<Object> actualTimeStampAndDate = new ArrayList<>();
+
+		Row row = null;
+		int count = 0;
+		for (FileInputSplit split : inputSplits) {
+			rowOrcInputFormat.open(split);
+			while (!rowOrcInputFormat.reachedEnd()) {
+				row = rowOrcInputFormat.nextRecord(row);
+				count++;
+				if (count <= 10000) {
+					actualTimeStampAndDate.add(row.getField(0) + "," + row.getField(1));
+				}
+
+			}
+		}
+		Assert.assertEquals(count, 70000);
+		Assert.assertEquals(expectedTimeStampAndDate.size(), actualTimeStampAndDate.size());
+		Assert.assertEquals(expectedTimeStampAndDate.toString(), actualTimeStampAndDate.toString());
+
+	}
+
+	@Test
+	public void testDecimal() throws IOException{
+
+		URL expectedDataURL = getClass().getClassLoader().getResource("decimal.dat");
+		List<String> expectedDecimal = Files.readAllLines(Paths.get(expectedDataURL.getPath()));
+
+		URL testInputURL = getClass().getClassLoader().getResource("decimal.orc");
+		assert(testInputURL != null);
+		String path = testInputURL.getPath();
+		String schema = "struct<_col0:decimal(10,5)>";
+		rowOrcInputFormat = new RowOrcInputFormat(path, schema, new Configuration());
+		rowOrcInputFormat.openInputFormat();
+
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		List<Object> actualDecimal = new ArrayList<>();
+
+		Row row = null;
+		for (FileInputSplit split : inputSplits) {
+			rowOrcInputFormat.open(split);
+			while (!rowOrcInputFormat.reachedEnd()) {
+				row = rowOrcInputFormat.nextRecord(row);
+				actualDecimal.add(row.getField(0));
+			}
+		}
+
+		Assert.assertEquals(expectedDecimal.size(), actualDecimal.size());
+		Assert.assertEquals(expectedDecimal.toString(), actualDecimal.toString());
+
+	}
+
+	@Test
+	public void testEmptyFile() throws IOException{
+
+		URL testInputURL = getClass().getClassLoader().getResource("TestOrcFile.emptyFile.orc");
+		assert(testInputURL != null);
+		String path = testInputURL.getPath();
+
+		rowOrcInputFormat = new RowOrcInputFormat(path, TEST1_SCHEMA, new Configuration());
+		rowOrcInputFormat.openInputFormat();
+
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		Row row = new Row(1);
+		int count = 0;
+		for (FileInputSplit split : inputSplits) {
+			rowOrcInputFormat.open(split);
+			while (!rowOrcInputFormat.reachedEnd()) {
+				row = rowOrcInputFormat.nextRecord(row);
+				count++;
+			}
+		}
+
+		Assert.assertEquals(count, 0);
+	}
+
+	@Test
+	public void testLargeFile() throws IOException{
+
+		URL testInputURL = getClass().getClassLoader().getResource("demo-11-none.orc");
+		assert(testInputURL != null);
+		String path = testInputURL.getPath();
+		String schema = "struct<_col0:int,_col1:string,_col2:string,_col3:string,_col4:int,"
+
+			"_col5:string,_col6:int,_col7:int,_col8:int>";
+
+		rowOrcInputFormat = new RowOrcInputFormat(path, schema, new Configuration());
+		rowOrcInputFormat.openInputFormat();
+
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		Row row = new Row(1);
+		int count = 0;
+		for (FileInputSplit split : inputSplits) {
+			rowOrcInputFormat.open(split);
+			while (!rowOrcInputFormat.reachedEnd()) {
+				row = rowOrcInputFormat.nextRecord(row);
+				count++;
+			}
+		}
+
+		Assert.assertEquals(count, 1920800);
+	}
+
+	@Test
+	public void testProducedType() throws IOException{
+
+		assert(test1URL != null);
+		rowOrcInputFormat = new RowOrcInputFormat(test1URL.getPath(), TEST1_SCHEMA, new Configuration());
+		rowOrcInputFormat.openInputFormat();
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		rowOrcInputFormat.open(inputSplits[0]);
+
+		TypeInformation<Row> type = rowOrcInputFormat.getProducedType();
+		Assert.assertEquals(type.toString(), "Row(boolean1: Boolean, byte1: Byte, short1: Short,
int1: Integer," +
+			" long1: Long, float1: Float, double1: Double, bytes1: byte[], string1: String," +
+			" middle: Row(list: ObjectArrayTypeInfo<Row(int1: Integer, string1: String)>),"
+
+			" list: ObjectArrayTypeInfo<Row(int1: Integer, string1: String)>," +
+			" map: Map<String, Row(int1: Integer, string1: String)>)");
+
+	}
+
+	@Test
+	public void testProducedTypeWithProjection() throws IOException{
+
+		assert(test1URL != null);
+		rowOrcInputFormat = new RowOrcInputFormat(test1URL.getPath(), TEST1_SCHEMA, new Configuration());
+		int[] projectionMask = {9, 10, 11};
+		rowOrcInputFormat.setFieldMapping(projectionMask);
+		rowOrcInputFormat.openInputFormat();
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		rowOrcInputFormat.open(inputSplits[0]);
+
+		TypeInformation<Row> type = rowOrcInputFormat.getProducedType();
+		Assert.assertEquals(type.toString(), "Row(middle: Row(list: ObjectArrayTypeInfo<Row(int1:
Integer, string1: String)>)," +
+			" list: ObjectArrayTypeInfo<Row(int1: Integer, string1: String)>," +
+			" map: Map<String, Row(int1: Integer, string1: String)>)");
+
+	}
+
+	@Test
+	public void testLongList() throws Exception {
+
+		URL testInputURL = getClass().getClassLoader().getResource("TestOrcFile.listlong.orc");
+		assert(testInputURL != null);
+		String path = testInputURL.getPath();
+		String schema = "struct<mylist1:array<bigint>>";
+
+		rowOrcInputFormat = new RowOrcInputFormat(path, schema, new Configuration());
+
+		rowOrcInputFormat.openInputFormat();
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		Row row = null;
+		long count = 0;
+		for (FileInputSplit split : inputSplits) {
+			rowOrcInputFormat.open(split);
+			while (!rowOrcInputFormat.reachedEnd()) {
+				row = rowOrcInputFormat.nextRecord(row);
+				Assert.assertEquals(row.getArity(), 1);
+				Object object = row.getField(0);
+				long[] l = (long[]) object;
+
+				Assert.assertEquals(l.length, 2);
+				if (count < 50) {
+					Assert.assertArrayEquals(l, new long[]{count, count + 1});
+				}
+				else {
+					Assert.assertArrayEquals(l, new long[]{0L, 0L});
+				}
+				count = count + 2;
+			}
+		}
+		Assert.assertEquals(count, 100);
+	}
+
+	@Test
+	public void testStringList() throws Exception {
+
+		URL testInputURL = getClass().getClassLoader().getResource("TestOrcFile.liststring.orc");
+		assert(testInputURL != null);
+		String path = testInputURL.getPath();
+		String schema = "struct<mylist1:array<string>>";
+
+		rowOrcInputFormat = new RowOrcInputFormat(path, schema, new Configuration());
+
+		rowOrcInputFormat.openInputFormat();
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		Row row = null;
+		long count = 0;
+		for (FileInputSplit split : inputSplits) {
+			rowOrcInputFormat.open(split);
+			while (!rowOrcInputFormat.reachedEnd()) {
+				row = rowOrcInputFormat.nextRecord(row);
+				Assert.assertEquals(row.getArity(), 1);
+				Object object = row.getField(0);
+				String[] l = (String[]) object;
+
+				Assert.assertEquals(l.length, 2);
+				Assert.assertArrayEquals(l, new String[]{"hello" + count, "hello" + (count + 1) });
+				count = count + 2;
+			}
+		}
+		Assert.assertEquals(count, 200);
+	}
+
+	@Test
+	public void testListOfListOfStructOfLong() throws Exception {
+		URL testInputURL = getClass().getClassLoader().getResource("TestOrcFile.listliststructlong.orc");
+		assert(testInputURL != null);
+		String path = testInputURL.getPath();
+		String schema = "struct<mylist1:array<array<struct<mylong1:bigint>>>>";
+
+		rowOrcInputFormat = new RowOrcInputFormat(path, schema, new Configuration());
+
+		rowOrcInputFormat.openInputFormat();
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(1);
+
+		Assert.assertEquals(inputSplits.length, 1);
+
+		Row row = null;
+		long count = 0;
+		for (FileInputSplit split : inputSplits) {
+			rowOrcInputFormat.open(split);
+			while (!rowOrcInputFormat.reachedEnd()) {
+
+				row = rowOrcInputFormat.nextRecord(row);
+				Assert.assertEquals(row.getArity(), 1);
+
+				Object[] objects = (Object[]) row.getField(0);
+				Assert.assertEquals(objects.length, 1);
+
+				Object[] objects1 = (Object[]) objects[0];
+				Assert.assertEquals(objects1.length, 1);
+
+				Row[] nestedRows = Arrays.copyOf(objects1, objects1.length, Row[].class);
+				Assert.assertEquals(nestedRows.length, 1);
+
+				Assert.assertEquals(nestedRows[0].getArity(), 1);
+
+				Assert.assertEquals(nestedRows[0].getField(0), count);
+
+				count++;
+			}
+		}
+		Assert.assertEquals(count, 100);
+	}
+
+	@Test
+	public void testSplit() throws IOException{
+
+		URL testInputURL = getClass().getClassLoader().getResource("demo-11-none.orc");
+		assert(testInputURL != null);
+		String path = testInputURL.getPath();
+		String schema = "struct<_col0:int,_col1:string,_col2:string,_col3:string,_col4:int,"
+
+			"_col5:string,_col6:int,_col7:int,_col8:int>";
+
+		rowOrcInputFormat = new RowOrcInputFormat(path, schema, new Configuration());
+		rowOrcInputFormat.openInputFormat();
+
+		FileInputSplit[] inputSplits = rowOrcInputFormat.createInputSplits(10);
+
+		Assert.assertEquals(inputSplits.length, 10);
+
+		Row row = null;
+		int countTotalRecords = 0;
+		for (FileInputSplit split : inputSplits) {
+			rowOrcInputFormat.open(split);
+			int countSplitRecords = 0;
+			while (!rowOrcInputFormat.reachedEnd()) {
+				row = rowOrcInputFormat.nextRecord(row);
+				countSplitRecords++;
+			}
+			Assert.assertNotEquals(countSplitRecords, 1920800);
+			countTotalRecords += countSplitRecords;
+		}
+
+		Assert.assertEquals(countTotalRecords, 1920800);
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/edbf8c9e/flink-connectors/flink-orc/src/test/resources/TestOrcFile.emptyFile.orc
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/test/resources/TestOrcFile.emptyFile.orc b/flink-connectors/flink-orc/src/test/resources/TestOrcFile.emptyFile.orc
new file mode 100644
index 0000000..ecdadcb
Binary files /dev/null and b/flink-connectors/flink-orc/src/test/resources/TestOrcFile.emptyFile.orc
differ

http://git-wip-us.apache.org/repos/asf/flink/blob/edbf8c9e/flink-connectors/flink-orc/src/test/resources/TestOrcFile.listliststructlong.orc
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/test/resources/TestOrcFile.listliststructlong.orc
b/flink-connectors/flink-orc/src/test/resources/TestOrcFile.listliststructlong.orc
new file mode 100644
index 0000000..0f3f9c8
Binary files /dev/null and b/flink-connectors/flink-orc/src/test/resources/TestOrcFile.listliststructlong.orc
differ

http://git-wip-us.apache.org/repos/asf/flink/blob/edbf8c9e/flink-connectors/flink-orc/src/test/resources/TestOrcFile.listlong.orc
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/test/resources/TestOrcFile.listlong.orc b/flink-connectors/flink-orc/src/test/resources/TestOrcFile.listlong.orc
new file mode 100644
index 0000000..648ea18
Binary files /dev/null and b/flink-connectors/flink-orc/src/test/resources/TestOrcFile.listlong.orc
differ

http://git-wip-us.apache.org/repos/asf/flink/blob/edbf8c9e/flink-connectors/flink-orc/src/test/resources/TestOrcFile.liststring.orc
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/test/resources/TestOrcFile.liststring.orc b/flink-connectors/flink-orc/src/test/resources/TestOrcFile.liststring.orc
new file mode 100644
index 0000000..75a5f2a
Binary files /dev/null and b/flink-connectors/flink-orc/src/test/resources/TestOrcFile.liststring.orc
differ

http://git-wip-us.apache.org/repos/asf/flink/blob/edbf8c9e/flink-connectors/flink-orc/src/test/resources/TestOrcFile.test1.orc
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/test/resources/TestOrcFile.test1.orc b/flink-connectors/flink-orc/src/test/resources/TestOrcFile.test1.orc
new file mode 100644
index 0000000..4fb0bef
Binary files /dev/null and b/flink-connectors/flink-orc/src/test/resources/TestOrcFile.test1.orc
differ


Mime
View raw message