You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by fhueske <gi...@git.apache.org> on 2017/11/21 21:11:49 UTC
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
GitHub user fhueske opened a pull request:
https://github.com/apache/flink/pull/5043
[FLINK-2170] [connectors] Add OrcRowInputFormat and OrcTableSource.
## What is the purpose of the change
* Adds `OrcRowInputFormat` to read [ORC files](https://orc.apache.org) as `DataSet<Row>`. The input format supports projection and filter push-down.
* Adds `OrcTableSource` to read [ORC files](https://orc.apache.org) as a `Table` in a batch Table API or SQL query. The table source supports projection and filter push-down.
## Brief change log
* Creates a new module `flink-connectors/flink-orc`
* Add `OrcRowInputFormat`
* Add `OrcTableSource`
* Add tests for input format and table source
* Adjust cost model of batch table scans to favor table sources with pushed-down filters over those without pushed-down filters.
* Add static method to `RowTypeInfo` to project on fields.
* Improve translation of literals in `RexProgramExtractor`
## Verifying this change
* `OrcRowInputFormatTest` verifies
* Correct configuration of ORC readers.
* Results when reading ORC files
* Schema evolution support
* Computation of split boundaries
* `OrcTableSourceTest` verifies
* Correct implementation of TableSource interface methods
* Correct configuration of `OrcRowInputFormat` for test queries (predicate and filter push-down)
* `OrcTableSourceITCase` runs end-to-end tests with SQL queries.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): **yes**, adds a new Maven module `flink-orc` with a dependency on `org.apache.orc/orc-core`
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **no**
- The serializers: **no**
- The runtime per-record code paths (performance sensitive): **no**
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: **no**
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **yes**
- If yes, how is the feature documented? **yes**, documentation for `RowTableSource` was added.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/fhueske/flink table-ORC
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5043.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5043
----
commit 2f524dfa0c4f8468691151925a622ba7fee55f0f
Author: uybhatti <uy...@gmail.com>
Date: 2017-03-03T22:55:22Z
[FLINK-2170] [connectors] Add OrcRowInputFormat and OrcTableSource.
commit d80506e3785268f541457a69ade3118c634cf7e7
Author: Fabian Hueske <fh...@apache.org>
Date: 2017-11-13T13:54:54Z
[FLINK-2170] [connectors] Add OrcRowInputFormat and OrcTableSource.
----
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152518464
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcRowInputFormat.java ---
@@ -0,0 +1,747 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.StripeInformation;
+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 java.io.Serializable;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.orc.OrcUtils.fillRows;
+
+/**
+ * InputFormat to read ORC files.
+ */
+public class OrcRowInputFormat extends FileInputFormat<Row> implements ResultTypeQueryable<Row> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrcRowInputFormat.class);
+ // the number of rows read in a batch
+ private static final int DEFAULT_BATCH_SIZE = 1000;
+
+ // the number of fields rows to read in a batch
+ private int batchSize;
+ // the configuration to read with
+ private Configuration conf;
+ // the schema of the ORC files to read
+ private TypeDescription schema;
+
+ // the fields of the ORC schema that the returned Rows are composed of.
+ private int[] selectedFields;
+ // the type information of the Rows returned by this InputFormat.
+ private transient RowTypeInfo rowType;
+
+ // the ORC reader
+ private transient RecordReader orcRowsReader;
+ // the vectorized row data to be read in a batch
+ private transient VectorizedRowBatch rowBatch;
+ // the vector of rows that is read in a batch
+ private transient Row[] rows;
+
+ // the number of rows in the current batch
+ private transient int rowsInBatch;
+ // the index of the next row to return
+ private transient int nextRow;
+
+ private ArrayList<Predicate> conjunctPredicates = new ArrayList<>();
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig, int batchSize) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, batchSize);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param orcSchema The schema of the ORC files as ORC TypeDescription.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize) {
+ super(new Path(path));
+
+ // configure OrcInputFormat
+ this.schema = orcSchema;
+ this.rowType = (RowTypeInfo) OrcUtils.schemaToTypeInfo(schema);
+ this.conf = orcConfig;
+ this.batchSize = batchSize;
+
+ // set default selection mask, i.e., all fields.
+ this.selectedFields = new int[this.schema.getChildren().size()];
+ for (int i = 0; i < selectedFields.length; i++) {
+ this.selectedFields[i] = i;
+ }
+ }
+
+ /**
+ * Adds a filter predicate to reduce the number of rows to be returned by the input format.
+ * Multiple conjunctive predicates can be added by calling this method multiple times.
+ *
+ * <p>Note: Predicates can significantly reduce the amount of data that is read.
+ * However, the OrcRowInputFormat does not guarantee that all returned rows qualitfy the
--- End diff --
qualify
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152526459
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcTableSource.java ---
@@ -0,0 +1,429 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
+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.RowTypeInfo;
+import org.apache.flink.orc.OrcRowInputFormat.Predicate;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.expressions.Attribute;
+import org.apache.flink.table.expressions.BinaryComparison;
+import org.apache.flink.table.expressions.EqualTo;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.GreaterThan;
+import org.apache.flink.table.expressions.GreaterThanOrEqual;
+import org.apache.flink.table.expressions.IsFalse;
+import org.apache.flink.table.expressions.IsNotNull;
+import org.apache.flink.table.expressions.IsNull;
+import org.apache.flink.table.expressions.IsTrue;
+import org.apache.flink.table.expressions.LessThan;
+import org.apache.flink.table.expressions.LessThanOrEqual;
+import org.apache.flink.table.expressions.Literal;
+import org.apache.flink.table.expressions.Not;
+import org.apache.flink.table.expressions.NotEqualTo;
+import org.apache.flink.table.expressions.Or;
+import org.apache.flink.table.expressions.UnaryExpression;
+import org.apache.flink.table.sources.BatchTableSource;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.types.Row;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.orc.TypeDescription;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * A TableSource to read ORC files.
+ *
+ * <p>The {@link OrcTableSource} supports projection and filter push-down.</p>
+ *
+ * <p>An {@link OrcTableSource} is used as shown in the example below.
+ *
+ * <pre>
+ * {@code
+ * String path = "file:///my/data/file.orc";
+ * String schema = "struct<col1:boolean,col2:tinyint,col3:smallint,col4:int>"
+ * OrcTableSource orcSrc = new OrcTableSource(path, schema);
+ * tEnv.registerTableSource("orcTable", orcSrc);
+ * Table res = tableEnv.sql("SELECT * FROM orcTable");
+ * }
+ * </pre>
+ */
+public class OrcTableSource
+ implements BatchTableSource<Row>, ProjectableTableSource<Row>, FilterableTableSource<Row> {
+
+ private static final int DEFAULT_BATCH_SIZE = 1024;
+
+ // path to read ORC files from
+ private final String path;
+ // schema of the ORC file
+ private final TypeDescription orcSchema;
+ // the schema of the Table
+ private final TableSchema tableSchema;
+ // the configuration to read the file
+ private final Configuration orcConfig;
+ // the number of rows to read in a batch
+ private final int batchSize;
+
+ // type information of the data returned by the InputFormat
+ private final RowTypeInfo typeInfo;
+ // list of selected ORC fields to return
+ private final int[] selectedFields;
+ // list of predicates to apply
+ private final Predicate[] predicates;
+
+ /**
+ * Creates an OrcTableSouce with empty configuration.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as String.
+ */
+ public OrcTableSource(String path, String orcSchema) {
+ this(path, orcSchema, new Configuration(), DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcTableSource from an ORC schema string.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files.
+ */
+ public OrcTableSource(String path, String orcSchema, Configuration orcConfig) {
+ this(path, TypeDescription.fromString(orcSchema), orcConfig, DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcTableSource from an ORC schema string.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files.
+ * @param batchSize The number of Rows to read in a batch, default is 1000.
+ */
+ public OrcTableSource(String path, String orcSchema, Configuration orcConfig, int batchSize) {
+ this(path, TypeDescription.fromString(orcSchema), orcConfig, batchSize);
+ }
+
+ /**
+ * Creates an OrcTableSouce from an ORC TypeDescription.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as TypeDescription.
+ * @param orcConfig The configuration to read the ORC files.
+ * @param batchSize The number of Rows to read in a batch, default is 1000.
+ */
+ public OrcTableSource(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize) {
+ this(path, orcSchema, orcConfig, batchSize, null, null);
+ }
+
+ private OrcTableSource(String path, TypeDescription orcSchema, Configuration orcConfig,
+ int batchSize, int[] selectedFields, Predicate[] predicates) {
+
+ this.path = path;
--- End diff --
Add some validation of the parameters?
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152503422
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcRowInputFormat.java ---
@@ -0,0 +1,747 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.StripeInformation;
+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 java.io.Serializable;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.orc.OrcUtils.fillRows;
+
+/**
+ * InputFormat to read ORC files.
+ */
+public class OrcRowInputFormat extends FileInputFormat<Row> implements ResultTypeQueryable<Row> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrcRowInputFormat.class);
+ // the number of rows read in a batch
+ private static final int DEFAULT_BATCH_SIZE = 1000;
+
+ // the number of fields rows to read in a batch
+ private int batchSize;
+ // the configuration to read with
+ private Configuration conf;
+ // the schema of the ORC files to read
+ private TypeDescription schema;
+
+ // the fields of the ORC schema that the returned Rows are composed of.
+ private int[] selectedFields;
+ // the type information of the Rows returned by this InputFormat.
+ private transient RowTypeInfo rowType;
+
+ // the ORC reader
+ private transient RecordReader orcRowsReader;
+ // the vectorized row data to be read in a batch
+ private transient VectorizedRowBatch rowBatch;
+ // the vector of rows that is read in a batch
+ private transient Row[] rows;
+
+ // the number of rows in the current batch
+ private transient int rowsInBatch;
+ // the index of the next row to return
+ private transient int nextRow;
+
+ private ArrayList<Predicate> conjunctPredicates = new ArrayList<>();
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig, int batchSize) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, batchSize);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param orcSchema The schema of the ORC files as ORC TypeDescription.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize) {
+ super(new Path(path));
+
+ // configure OrcInputFormat
--- End diff --
`OrcRowInputFormat`
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152508692
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcUtils.java ---
@@ -0,0 +1,1511 @@
+/*
+ * 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.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+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.types.Row;
+
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+
+import org.apache.orc.TypeDescription;
+
+import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.function.DoubleFunction;
+import java.util.function.IntFunction;
+import java.util.function.LongFunction;
+
+/**
+ * A class that provides utility methods for orc file reading.
+ */
+class OrcUtils {
+
+ private static final long MILLIS_PER_DAY = 86400000; // = 24 * 60 * 60 * 1000
+ private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
+ /**
+ * Converts an ORC schema to a Flink TypeInformation.
+ *
+ * @param schema The ORC schema.
+ * @return The TypeInformation that corresponds to the ORC schema.
+ */
+ static TypeInformation schemaToTypeInfo(TypeDescription schema) {
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ return BasicTypeInfo.BOOLEAN_TYPE_INFO;
+ case BYTE:
+ return BasicTypeInfo.BYTE_TYPE_INFO;
+ case SHORT:
+ return BasicTypeInfo.SHORT_TYPE_INFO;
+ case INT:
+ return BasicTypeInfo.INT_TYPE_INFO;
+ case LONG:
+ return BasicTypeInfo.LONG_TYPE_INFO;
+ case FLOAT:
+ return BasicTypeInfo.FLOAT_TYPE_INFO;
+ case DOUBLE:
+ return BasicTypeInfo.DOUBLE_TYPE_INFO;
+ case DECIMAL:
+ return BasicTypeInfo.BIG_DEC_TYPE_INFO;
+ case STRING:
+ case CHAR:
+ case VARCHAR:
+ return BasicTypeInfo.STRING_TYPE_INFO;
+ case DATE:
+ return SqlTimeTypeInfo.DATE;
+ case TIMESTAMP:
+ return SqlTimeTypeInfo.TIMESTAMP;
+ case BINARY:
+ return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO;
+ case STRUCT:
+ List<TypeDescription> fieldSchemas = schema.getChildren();
+ TypeInformation[] fieldTypes = new TypeInformation[fieldSchemas.size()];
+ for (int i = 0; i < fieldSchemas.size(); i++) {
+ fieldTypes[i] = schemaToTypeInfo(fieldSchemas.get(i));
+ }
+ String[] fieldNames = schema.getFieldNames().toArray(new String[]{});
+ return new RowTypeInfo(fieldTypes, fieldNames);
+ case LIST:
+ TypeDescription elementSchema = schema.getChildren().get(0);
+ TypeInformation<?> elementType = schemaToTypeInfo(elementSchema);
+ // arrays of primitive types are handled as object arrays to support null values
+ return ObjectArrayTypeInfo.getInfoFor(elementType);
+ case MAP:
+ TypeDescription keySchema = schema.getChildren().get(0);
+ TypeDescription valSchema = schema.getChildren().get(1);
+ TypeInformation<?> keyType = schemaToTypeInfo(keySchema);
+ TypeInformation<?> valType = schemaToTypeInfo(valSchema);
+ return new MapTypeInfo<>(keyType, valType);
+ case UNION:
+ throw new UnsupportedOperationException("UNION type is not supported yet.");
+ default:
+ throw new IllegalArgumentException("Unknown type " + schema);
+ }
+ }
+
+ /**
+ * Fills an ORC batch into an array of Row.
+ *
+ * @param rows The batch of rows need to be filled.
+ * @param schema The schema of the ORC data.
+ * @param batch The ORC data.
+ * @param selectedFields The list of selected ORC fields.
+ * @return The number of rows that were filled.
+ */
+ static int fillRows(Row[] rows, TypeDescription schema, VectorizedRowBatch batch, int[] selectedFields) {
+
+ int rowsToRead = Math.min((int) batch.count(), rows.length);
+
+ List<TypeDescription> fieldTypes = schema.getChildren();
+ // read each selected field
+ for (int rowIdx = 0; rowIdx < selectedFields.length; rowIdx++) {
+ int orcIdx = selectedFields[rowIdx];
+ readField(rows, rowIdx, fieldTypes.get(orcIdx), batch.cols[orcIdx], null, rowsToRead);
+ }
+ return rowsToRead;
+ }
+
+ /**
+ * Reads a vector of data into an array of objects.
+ *
+ * @param vals The array that needs to be filled.
+ * @param fieldIdx If the vals array is an array of Row, the index of the field that needs to be filled.
+ * Otherwise a -1 must be passed and the data is directly filled into the array.
+ * @param schema The schema of the vector to read.
+ * @param vector The vector to read.
+ * @param lengthVector If the vector is of type List or Map, the number of sub-elements to read for each field. Otherwise, it must be null.
+ * @param childCount The number of vector entries to read.
+ */
+ private static void readField(Object[] vals, int fieldIdx, TypeDescription schema, ColumnVector vector, long[] lengthVector, int childCount) {
+
+ // check the type of the vector to decide how to read it.
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ }
+ break;
+ case BYTE:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ }
+ break;
+ case SHORT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ }
+ break;
+ case INT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ }
+ break;
+ case LONG:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ }
+ break;
+ case FLOAT:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ }
+ break;
+ case DOUBLE:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ }
+ break;
+ case CHAR:
+ case VARCHAR:
+ case STRING:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DATE:
+ if (vector.noNulls) {
+ readNonNullLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ } else {
+ readLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case TIMESTAMP:
+ if (vector.noNulls) {
+ readNonNullTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ } else {
+ readTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case BINARY:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DECIMAL:
+ if (vector.noNulls) {
+ readNonNullDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ else {
+ readDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case STRUCT:
+ if (vector.noNulls) {
+ readNonNullStructColumn(vals, fieldIdx, (StructColumnVector) vector, schema, lengthVector, childCount);
+ } else {
+ readStructColumn(vals, fieldIdx, (StructColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case LIST:
+ if (vector.noNulls) {
+ readNonNullListColumn(vals, fieldIdx, (ListColumnVector) vector, schema, lengthVector, childCount);
+ }
+ else {
+ readListColumn(vals, fieldIdx, (ListColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case MAP:
+ if (vector.noNulls) {
+ readNonNullMapColumn(vals, fieldIdx, (MapColumnVector) vector, schema, lengthVector, childCount);
+ }
+ else {
+ readMapColumn(vals, fieldIdx, (MapColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case UNION:
--- End diff --
Add `IntervalDayTimeColumnVector` also not supported yet? It is not in `TypeDescriptor` but exists as ColumnVector. Do you know about it?
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152535711
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcRowInputFormat.java ---
@@ -0,0 +1,747 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.StripeInformation;
+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 java.io.Serializable;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.orc.OrcUtils.fillRows;
+
+/**
+ * InputFormat to read ORC files.
+ */
+public class OrcRowInputFormat extends FileInputFormat<Row> implements ResultTypeQueryable<Row> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrcRowInputFormat.class);
+ // the number of rows read in a batch
+ private static final int DEFAULT_BATCH_SIZE = 1000;
+
+ // the number of fields rows to read in a batch
+ private int batchSize;
+ // the configuration to read with
+ private Configuration conf;
+ // the schema of the ORC files to read
+ private TypeDescription schema;
+
+ // the fields of the ORC schema that the returned Rows are composed of.
+ private int[] selectedFields;
+ // the type information of the Rows returned by this InputFormat.
+ private transient RowTypeInfo rowType;
+
+ // the ORC reader
+ private transient RecordReader orcRowsReader;
+ // the vectorized row data to be read in a batch
+ private transient VectorizedRowBatch rowBatch;
+ // the vector of rows that is read in a batch
+ private transient Row[] rows;
+
+ // the number of rows in the current batch
+ private transient int rowsInBatch;
+ // the index of the next row to return
+ private transient int nextRow;
+
+ private ArrayList<Predicate> conjunctPredicates = new ArrayList<>();
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig, int batchSize) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, batchSize);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param orcSchema The schema of the ORC files as ORC TypeDescription.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize) {
+ super(new Path(path));
+
+ // configure OrcInputFormat
+ this.schema = orcSchema;
+ this.rowType = (RowTypeInfo) OrcUtils.schemaToTypeInfo(schema);
+ this.conf = orcConfig;
+ this.batchSize = batchSize;
+
+ // set default selection mask, i.e., all fields.
+ this.selectedFields = new int[this.schema.getChildren().size()];
+ for (int i = 0; i < selectedFields.length; i++) {
+ this.selectedFields[i] = i;
+ }
+ }
+
+ /**
+ * Adds a filter predicate to reduce the number of rows to be returned by the input format.
+ * Multiple conjunctive predicates can be added by calling this method multiple times.
+ *
+ * <p>Note: Predicates can significantly reduce the amount of data that is read.
+ * However, the OrcRowInputFormat does not guarantee that all returned rows qualitfy the
+ * predicates. Moreover, predicates are only applied if the referenced field is among the
+ * selected fields.</p>
+ *
+ * @param predicate The filter predicate.
+ */
+ public void addPredicate(Predicate predicate) {
+ // validate
+ validatePredicate(predicate);
+ // add predicate
+ this.conjunctPredicates.add(predicate);
+ }
+
+ private void validatePredicate(Predicate pred) {
+ if (pred instanceof ColumnPredicate) {
+ // check column name
+ String colName = ((ColumnPredicate) pred).columnName;
+ if (!this.schema.getFieldNames().contains(colName)) {
+ throw new IllegalArgumentException("Predicate cannot be applied. " +
+ "Column '" + colName + "' does not exist in ORC schema.");
+ }
+ } else if (pred instanceof Not) {
+ validatePredicate(((Not) pred).child());
+ } else if (pred instanceof Or) {
+ for (Predicate p : ((Or) pred).children()) {
+ validatePredicate(p);
+ }
+ }
+ }
+
+ /**
+ * Selects the fields from the ORC schema that are returned by InputFormat.
+ *
+ * @param selectedFields The indices of the fields of the ORC schema that are returned by the InputFormat.
+ */
+ public void selectFields(int... selectedFields) {
+ // set field mapping
+ this.selectedFields = selectedFields;
+ // adapt result type
+ this.rowType = RowTypeInfo.projectFields(this.rowType, selectedFields);
+ }
+
+ /**
+ * Computes the ORC projection mask of the fields to include from the selected fields.rowOrcInputFormat.nextRecord(null).
+ *
+ * @return The ORC projection mask.
+ */
+ private boolean[] computeProjectionMask() {
+ // mask with all fields of the schema
+ boolean[] projectionMask = new boolean[schema.getMaximumId() + 1];
+ // for each selected field
+ for (int inIdx : selectedFields) {
+ // set all nested fields of a selected field to true
+ 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();
+ // create and initialize the row batch
+ this.rows = new Row[batchSize];
+ for (int i = 0; i < batchSize; i++) {
+ rows[i] = new Row(selectedFields.length);
+ }
+ }
+
+ @Override
+ public void open(FileInputSplit fileSplit) throws IOException {
+
+ LOG.debug("Opening ORC file {}", fileSplit.getPath());
+
+ // open ORC file and create reader
+ org.apache.hadoop.fs.Path hPath = new org.apache.hadoop.fs.Path(fileSplit.getPath().getPath());
+ Reader orcReader = OrcFile.createReader(hPath, OrcFile.readerOptions(conf));
+
+ // get offset and length for the stripes that start in the split
+ Tuple2<Long, Long> offsetAndLength = getOffsetAndLengthForSplit(fileSplit, getStripes(orcReader));
+
+ // create ORC row reader configuration
+ Reader.Options options = getOptions(orcReader)
+ .schema(schema)
+ .range(offsetAndLength.f0, offsetAndLength.f1)
+ .useZeroCopy(OrcConf.USE_ZEROCOPY.getBoolean(conf))
+ .skipCorruptRecords(OrcConf.SKIP_CORRUPT_DATA.getBoolean(conf))
+ .tolerateMissingSchema(OrcConf.TOLERATE_MISSING_SCHEMA.getBoolean(conf));
+
+ // configure filters
+ if (!conjunctPredicates.isEmpty()) {
+ SearchArgument.Builder b = SearchArgumentFactory.newBuilder();
+ b = b.startAnd();
+ for (Predicate predicate : conjunctPredicates) {
+ predicate.add(b);
+ }
+ b = b.end();
+ options.searchArgument(b.build(), new String[]{});
+ }
+
+ // configure selected fields
+ options.include(computeProjectionMask());
+
+ // create ORC row reader
+ this.orcRowsReader = orcReader.rows(options);
+
+ // assign ids
+ this.schema.getId();
+ // create row batch
+ this.rowBatch = schema.createRowBatch(batchSize);
+ rowsInBatch = 0;
+ nextRow = 0;
+ }
+
+ @VisibleForTesting
+ Reader.Options getOptions(Reader orcReader) {
+ return orcReader.options();
+ }
+
+ @VisibleForTesting
+ List<StripeInformation> getStripes(Reader orcReader) {
+ return orcReader.getStripes();
+ }
+
+ private Tuple2<Long, Long> getOffsetAndLengthForSplit(FileInputSplit split, List<StripeInformation> stripes) {
+ long splitStart = split.getStart();
+ long splitEnd = splitStart + split.getLength();
+
+ long readStart = Long.MAX_VALUE;
+ long readEnd = Long.MIN_VALUE;
+
+ for (StripeInformation s : stripes) {
+ if (splitStart <= s.getOffset() && s.getOffset() < splitEnd) {
+ // stripe starts in split, so it is included
+ readStart = Math.min(readStart, s.getOffset());
+ readEnd = Math.max(readEnd, s.getOffset() + s.getLength());
+ }
+ }
+
+ if (readStart < Long.MAX_VALUE) {
+ // at least one split is included
+ return Tuple2.of(readStart, readEnd - readStart);
+ } else {
+ return Tuple2.of(0L, 0L);
+ }
+ }
+
+ @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();
+ }
+
+ /**
+ * Checks if there is at least one row left in the batch to return.
+ * If no more row are available, it reads another batch of rows.
+ *
+ * @return Returns true if there is one more row to return, false otherwise.
+ * @throws IOException throw if an exception happens while reading a batch.
+ */
+ private boolean ensureBatch() throws IOException {
+
+ if (nextRow >= rowsInBatch) {
+ // No more rows available in the Rows array.
+ nextRow = 0;
+ // Try to read the next batch if rows from the ORC file.
+ boolean moreRows = orcRowsReader.nextBatch(rowBatch);
+
+ if (moreRows) {
+ // Load the data into the Rows array.
+ rowsInBatch = fillRows(rows, schema, rowBatch, selectedFields);
+ }
+ return moreRows;
+ }
+ // there is at least one Row left in the Rows array.
+ return true;
+ }
+
+ @Override
+ public Row nextRecord(Row reuse) throws IOException {
+ // return the next row
+ return rows[this.nextRow++];
+ }
+
+ @Override
+ public TypeInformation<Row> getProducedType() {
+ return rowType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Custom serialization methods
+ // --------------------------------------------------------------------------------------------
+
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.writeInt(batchSize);
+ this.conf.write(out);
+ out.writeUTF(schema.toString());
+
+ out.writeInt(selectedFields.length);
+ for (int f : selectedFields) {
+ out.writeInt(f);
+ }
+
+ out.writeInt(conjunctPredicates.size());
+ for (Predicate p : conjunctPredicates) {
+ out.writeObject(p);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ batchSize = in.readInt();
+ org.apache.hadoop.conf.Configuration configuration = new org.apache.hadoop.conf.Configuration();
+ configuration.readFields(in);
+
+ if (this.conf == null) {
+ this.conf = configuration;
+ }
+ this.schema = TypeDescription.fromString(in.readUTF());
+
+ this.selectedFields = new int[in.readInt()];
+ for (int i = 0; i < selectedFields.length; i++) {
+ this.selectedFields[i] = in.readInt();
+ }
+
+ this.conjunctPredicates = new ArrayList<>();
+ int numPreds = in.readInt();
+ for (int i = 0; i < numPreds; i++) {
+ conjunctPredicates.add((Predicate) in.readObject());
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Classes to define predicates
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A filter predicate that can be evaluated by the OrcRowInputFormat.
+ */
+ public abstract static class Predicate implements Serializable {
+ protected abstract SearchArgument.Builder add(SearchArgument.Builder builder);
+ }
+
+ abstract static class ColumnPredicate extends Predicate {
+ final String columnName;
+ final PredicateLeaf.Type literalType;
+
+ ColumnPredicate(String columnName, PredicateLeaf.Type literalType) {
+ this.columnName = columnName;
+ this.literalType = literalType;
+ }
+
+ Object castLiteral(Serializable literal) {
+
+ switch (literalType) {
+ case LONG:
+ if (literal instanceof Byte) {
+ return new Long((Byte) literal);
+ } else if (literal instanceof Short) {
+ return new Long((Short) literal);
+ } else if (literal instanceof Integer) {
+ return new Long((Integer) literal);
+ } else if (literal instanceof Long) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a LONG column requires an integer " +
+ "literal, i.e., Byte, Short, Integer, or Long.");
+ }
+ case FLOAT:
+ if (literal instanceof Float) {
+ return new Double((Float) literal);
+ } else if (literal instanceof Double) {
+ return literal;
+ } else if (literal instanceof BigDecimal) {
+ return ((BigDecimal) literal).doubleValue();
+ } else {
+ throw new IllegalArgumentException("A predicate on a FLOAT column requires a floating " +
+ "literal, i.e., Float or Double.");
+ }
+ case STRING:
+ if (literal instanceof String) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a STRING column requires a floating " +
+ "literal, i.e., Float or Double.");
+ }
+ case BOOLEAN:
+ if (literal instanceof Boolean) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a BOOLEAN column requires a Boolean literal.");
+ }
+ case DATE:
+ if (literal instanceof Date) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a DATE column requires a java.sql.Date literal.");
+ }
+ case TIMESTAMP:
+ if (literal instanceof Timestamp) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a TIMESTAMP column requires a java.sql.Timestamp literal.");
+ }
+ case DECIMAL:
+ if (literal instanceof BigDecimal) {
+ return new HiveDecimalWritable(HiveDecimal.create((BigDecimal) literal));
+ } else {
+ throw new IllegalArgumentException("A predicate on a DECIMAL column requires a BigDecimal literal.");
+ }
+ default:
+ throw new IllegalArgumentException("Unknown literal type " + literalType);
+ }
+ }
+ }
+
+ abstract static class BinaryPredicate extends ColumnPredicate {
+ final Serializable literal;
+
+ BinaryPredicate(String columnName, PredicateLeaf.Type literalType, Serializable literal) {
+ super(columnName, literalType);
+ this.literal = literal;
+ }
+ }
+
+ /**
+ * An EQUALS predicate that can be evaluated by the OrcRowInputFormat.
+ */
+ public static class Equals extends BinaryPredicate {
+ /**
+ * Creates an EQUALS predicate.
+ *
+ * @param columnName The column to check.
+ * @param literalType The type of the literal.
+ * @param literal The literal value to check the column against.
+ */
+ public Equals(String columnName, PredicateLeaf.Type literalType, Serializable literal) {
+ super(columnName, literalType, literal);
+ }
+
+ @Override
+ protected SearchArgument.Builder add(SearchArgument.Builder builder) {
+ return builder.equals(columnName, literalType, castLiteral(literal));
+ }
+
+ @Override
+ public String toString() {
+ return columnName + " = " + literal;
+ }
+ }
+
+ /**
+ * An EQUALS predicate that can be evaluated with Null safety by the OrcRowInputFormat.
+ */
+ public static class NullSafeEquals extends BinaryPredicate {
+ /**
+ * Creates a null-safe EQUALS predicate.
+ *
+ * @param columnName The column to check.
+ * @param literalType The type of the literal.
+ * @param literal The literal value to check the column against.
+ */
+ public NullSafeEquals(String columnName, PredicateLeaf.Type literalType, Serializable literal) {
--- End diff --
Yes, but it's an interface class for users that are using the `OrcRowInputFormat` directly.
---
[GitHub] flink issue #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat and OrcTa...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/5043
Thanks for the review @twalthr!
I'll address your comments and will merge this PR.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152524790
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcRowInputFormat.java ---
@@ -0,0 +1,747 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.StripeInformation;
+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 java.io.Serializable;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.orc.OrcUtils.fillRows;
+
+/**
+ * InputFormat to read ORC files.
+ */
+public class OrcRowInputFormat extends FileInputFormat<Row> implements ResultTypeQueryable<Row> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrcRowInputFormat.class);
+ // the number of rows read in a batch
+ private static final int DEFAULT_BATCH_SIZE = 1000;
+
+ // the number of fields rows to read in a batch
+ private int batchSize;
+ // the configuration to read with
+ private Configuration conf;
+ // the schema of the ORC files to read
+ private TypeDescription schema;
+
+ // the fields of the ORC schema that the returned Rows are composed of.
+ private int[] selectedFields;
+ // the type information of the Rows returned by this InputFormat.
+ private transient RowTypeInfo rowType;
+
+ // the ORC reader
+ private transient RecordReader orcRowsReader;
+ // the vectorized row data to be read in a batch
+ private transient VectorizedRowBatch rowBatch;
+ // the vector of rows that is read in a batch
+ private transient Row[] rows;
+
+ // the number of rows in the current batch
+ private transient int rowsInBatch;
+ // the index of the next row to return
+ private transient int nextRow;
+
+ private ArrayList<Predicate> conjunctPredicates = new ArrayList<>();
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig, int batchSize) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, batchSize);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param orcSchema The schema of the ORC files as ORC TypeDescription.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize) {
+ super(new Path(path));
+
+ // configure OrcInputFormat
+ this.schema = orcSchema;
+ this.rowType = (RowTypeInfo) OrcUtils.schemaToTypeInfo(schema);
+ this.conf = orcConfig;
+ this.batchSize = batchSize;
+
+ // set default selection mask, i.e., all fields.
+ this.selectedFields = new int[this.schema.getChildren().size()];
+ for (int i = 0; i < selectedFields.length; i++) {
+ this.selectedFields[i] = i;
+ }
+ }
+
+ /**
+ * Adds a filter predicate to reduce the number of rows to be returned by the input format.
+ * Multiple conjunctive predicates can be added by calling this method multiple times.
+ *
+ * <p>Note: Predicates can significantly reduce the amount of data that is read.
+ * However, the OrcRowInputFormat does not guarantee that all returned rows qualitfy the
+ * predicates. Moreover, predicates are only applied if the referenced field is among the
+ * selected fields.</p>
+ *
+ * @param predicate The filter predicate.
+ */
+ public void addPredicate(Predicate predicate) {
+ // validate
+ validatePredicate(predicate);
+ // add predicate
+ this.conjunctPredicates.add(predicate);
+ }
+
+ private void validatePredicate(Predicate pred) {
+ if (pred instanceof ColumnPredicate) {
+ // check column name
+ String colName = ((ColumnPredicate) pred).columnName;
+ if (!this.schema.getFieldNames().contains(colName)) {
+ throw new IllegalArgumentException("Predicate cannot be applied. " +
+ "Column '" + colName + "' does not exist in ORC schema.");
+ }
+ } else if (pred instanceof Not) {
+ validatePredicate(((Not) pred).child());
+ } else if (pred instanceof Or) {
+ for (Predicate p : ((Or) pred).children()) {
+ validatePredicate(p);
+ }
+ }
+ }
+
+ /**
+ * Selects the fields from the ORC schema that are returned by InputFormat.
+ *
+ * @param selectedFields The indices of the fields of the ORC schema that are returned by the InputFormat.
+ */
+ public void selectFields(int... selectedFields) {
+ // set field mapping
+ this.selectedFields = selectedFields;
+ // adapt result type
+ this.rowType = RowTypeInfo.projectFields(this.rowType, selectedFields);
+ }
+
+ /**
+ * Computes the ORC projection mask of the fields to include from the selected fields.rowOrcInputFormat.nextRecord(null).
+ *
+ * @return The ORC projection mask.
+ */
+ private boolean[] computeProjectionMask() {
+ // mask with all fields of the schema
+ boolean[] projectionMask = new boolean[schema.getMaximumId() + 1];
+ // for each selected field
+ for (int inIdx : selectedFields) {
+ // set all nested fields of a selected field to true
+ 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();
+ // create and initialize the row batch
+ this.rows = new Row[batchSize];
+ for (int i = 0; i < batchSize; i++) {
+ rows[i] = new Row(selectedFields.length);
+ }
+ }
+
+ @Override
+ public void open(FileInputSplit fileSplit) throws IOException {
+
+ LOG.debug("Opening ORC file {}", fileSplit.getPath());
+
+ // open ORC file and create reader
+ org.apache.hadoop.fs.Path hPath = new org.apache.hadoop.fs.Path(fileSplit.getPath().getPath());
+ Reader orcReader = OrcFile.createReader(hPath, OrcFile.readerOptions(conf));
+
+ // get offset and length for the stripes that start in the split
+ Tuple2<Long, Long> offsetAndLength = getOffsetAndLengthForSplit(fileSplit, getStripes(orcReader));
+
+ // create ORC row reader configuration
+ Reader.Options options = getOptions(orcReader)
+ .schema(schema)
+ .range(offsetAndLength.f0, offsetAndLength.f1)
+ .useZeroCopy(OrcConf.USE_ZEROCOPY.getBoolean(conf))
+ .skipCorruptRecords(OrcConf.SKIP_CORRUPT_DATA.getBoolean(conf))
+ .tolerateMissingSchema(OrcConf.TOLERATE_MISSING_SCHEMA.getBoolean(conf));
+
+ // configure filters
+ if (!conjunctPredicates.isEmpty()) {
+ SearchArgument.Builder b = SearchArgumentFactory.newBuilder();
+ b = b.startAnd();
+ for (Predicate predicate : conjunctPredicates) {
+ predicate.add(b);
+ }
+ b = b.end();
+ options.searchArgument(b.build(), new String[]{});
+ }
+
+ // configure selected fields
+ options.include(computeProjectionMask());
+
+ // create ORC row reader
+ this.orcRowsReader = orcReader.rows(options);
+
+ // assign ids
+ this.schema.getId();
+ // create row batch
+ this.rowBatch = schema.createRowBatch(batchSize);
+ rowsInBatch = 0;
+ nextRow = 0;
+ }
+
+ @VisibleForTesting
+ Reader.Options getOptions(Reader orcReader) {
+ return orcReader.options();
+ }
+
+ @VisibleForTesting
+ List<StripeInformation> getStripes(Reader orcReader) {
+ return orcReader.getStripes();
+ }
+
+ private Tuple2<Long, Long> getOffsetAndLengthForSplit(FileInputSplit split, List<StripeInformation> stripes) {
+ long splitStart = split.getStart();
+ long splitEnd = splitStart + split.getLength();
+
+ long readStart = Long.MAX_VALUE;
+ long readEnd = Long.MIN_VALUE;
+
+ for (StripeInformation s : stripes) {
+ if (splitStart <= s.getOffset() && s.getOffset() < splitEnd) {
+ // stripe starts in split, so it is included
+ readStart = Math.min(readStart, s.getOffset());
+ readEnd = Math.max(readEnd, s.getOffset() + s.getLength());
+ }
+ }
+
+ if (readStart < Long.MAX_VALUE) {
+ // at least one split is included
+ return Tuple2.of(readStart, readEnd - readStart);
+ } else {
+ return Tuple2.of(0L, 0L);
+ }
+ }
+
+ @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();
+ }
+
+ /**
+ * Checks if there is at least one row left in the batch to return.
+ * If no more row are available, it reads another batch of rows.
+ *
+ * @return Returns true if there is one more row to return, false otherwise.
+ * @throws IOException throw if an exception happens while reading a batch.
+ */
+ private boolean ensureBatch() throws IOException {
+
+ if (nextRow >= rowsInBatch) {
+ // No more rows available in the Rows array.
+ nextRow = 0;
+ // Try to read the next batch if rows from the ORC file.
+ boolean moreRows = orcRowsReader.nextBatch(rowBatch);
+
+ if (moreRows) {
+ // Load the data into the Rows array.
+ rowsInBatch = fillRows(rows, schema, rowBatch, selectedFields);
+ }
+ return moreRows;
+ }
+ // there is at least one Row left in the Rows array.
+ return true;
+ }
+
+ @Override
+ public Row nextRecord(Row reuse) throws IOException {
+ // return the next row
+ return rows[this.nextRow++];
+ }
+
+ @Override
+ public TypeInformation<Row> getProducedType() {
+ return rowType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Custom serialization methods
+ // --------------------------------------------------------------------------------------------
+
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.writeInt(batchSize);
+ this.conf.write(out);
+ out.writeUTF(schema.toString());
+
+ out.writeInt(selectedFields.length);
+ for (int f : selectedFields) {
+ out.writeInt(f);
+ }
+
+ out.writeInt(conjunctPredicates.size());
+ for (Predicate p : conjunctPredicates) {
+ out.writeObject(p);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ batchSize = in.readInt();
+ org.apache.hadoop.conf.Configuration configuration = new org.apache.hadoop.conf.Configuration();
+ configuration.readFields(in);
+
+ if (this.conf == null) {
+ this.conf = configuration;
+ }
+ this.schema = TypeDescription.fromString(in.readUTF());
+
+ this.selectedFields = new int[in.readInt()];
+ for (int i = 0; i < selectedFields.length; i++) {
+ this.selectedFields[i] = in.readInt();
+ }
+
+ this.conjunctPredicates = new ArrayList<>();
+ int numPreds = in.readInt();
+ for (int i = 0; i < numPreds; i++) {
+ conjunctPredicates.add((Predicate) in.readObject());
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Classes to define predicates
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A filter predicate that can be evaluated by the OrcRowInputFormat.
+ */
+ public abstract static class Predicate implements Serializable {
+ protected abstract SearchArgument.Builder add(SearchArgument.Builder builder);
+ }
+
+ abstract static class ColumnPredicate extends Predicate {
+ final String columnName;
+ final PredicateLeaf.Type literalType;
+
+ ColumnPredicate(String columnName, PredicateLeaf.Type literalType) {
+ this.columnName = columnName;
+ this.literalType = literalType;
+ }
+
+ Object castLiteral(Serializable literal) {
+
+ switch (literalType) {
+ case LONG:
+ if (literal instanceof Byte) {
+ return new Long((Byte) literal);
+ } else if (literal instanceof Short) {
+ return new Long((Short) literal);
+ } else if (literal instanceof Integer) {
+ return new Long((Integer) literal);
+ } else if (literal instanceof Long) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a LONG column requires an integer " +
+ "literal, i.e., Byte, Short, Integer, or Long.");
+ }
+ case FLOAT:
+ if (literal instanceof Float) {
+ return new Double((Float) literal);
+ } else if (literal instanceof Double) {
+ return literal;
+ } else if (literal instanceof BigDecimal) {
+ return ((BigDecimal) literal).doubleValue();
+ } else {
+ throw new IllegalArgumentException("A predicate on a FLOAT column requires a floating " +
+ "literal, i.e., Float or Double.");
+ }
+ case STRING:
+ if (literal instanceof String) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a STRING column requires a floating " +
+ "literal, i.e., Float or Double.");
+ }
+ case BOOLEAN:
+ if (literal instanceof Boolean) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a BOOLEAN column requires a Boolean literal.");
+ }
+ case DATE:
+ if (literal instanceof Date) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a DATE column requires a java.sql.Date literal.");
+ }
+ case TIMESTAMP:
+ if (literal instanceof Timestamp) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a TIMESTAMP column requires a java.sql.Timestamp literal.");
+ }
+ case DECIMAL:
+ if (literal instanceof BigDecimal) {
+ return new HiveDecimalWritable(HiveDecimal.create((BigDecimal) literal));
+ } else {
+ throw new IllegalArgumentException("A predicate on a DECIMAL column requires a BigDecimal literal.");
+ }
+ default:
+ throw new IllegalArgumentException("Unknown literal type " + literalType);
+ }
+ }
+ }
+
+ abstract static class BinaryPredicate extends ColumnPredicate {
+ final Serializable literal;
+
+ BinaryPredicate(String columnName, PredicateLeaf.Type literalType, Serializable literal) {
+ super(columnName, literalType);
+ this.literal = literal;
+ }
+ }
+
+ /**
+ * An EQUALS predicate that can be evaluated by the OrcRowInputFormat.
+ */
+ public static class Equals extends BinaryPredicate {
+ /**
+ * Creates an EQUALS predicate.
+ *
+ * @param columnName The column to check.
+ * @param literalType The type of the literal.
+ * @param literal The literal value to check the column against.
+ */
+ public Equals(String columnName, PredicateLeaf.Type literalType, Serializable literal) {
+ super(columnName, literalType, literal);
+ }
+
+ @Override
+ protected SearchArgument.Builder add(SearchArgument.Builder builder) {
+ return builder.equals(columnName, literalType, castLiteral(literal));
+ }
+
+ @Override
+ public String toString() {
+ return columnName + " = " + literal;
+ }
+ }
+
+ /**
+ * An EQUALS predicate that can be evaluated with Null safety by the OrcRowInputFormat.
+ */
+ public static class NullSafeEquals extends BinaryPredicate {
+ /**
+ * Creates a null-safe EQUALS predicate.
+ *
+ * @param columnName The column to check.
+ * @param literalType The type of the literal.
+ * @param literal The literal value to check the column against.
+ */
+ public NullSafeEquals(String columnName, PredicateLeaf.Type literalType, Serializable literal) {
--- End diff --
This class is never used.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152501940
--- Diff: flink-connectors/flink-orc/pom.xml ---
@@ -0,0 +1,112 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connectors</artifactId>
+ <version>1.5-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-orc_${scala.binary.version}</artifactId>
+ <name>flink-orc</name>
+
+ <packaging>jar</packaging>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <!-- Projects depending on this project,
--- End diff --
Remove line break.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152580041
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcUtils.java ---
@@ -0,0 +1,1511 @@
+/*
+ * 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.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+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.types.Row;
+
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+
+import org.apache.orc.TypeDescription;
+
+import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.function.DoubleFunction;
+import java.util.function.IntFunction;
+import java.util.function.LongFunction;
+
+/**
+ * A class that provides utility methods for orc file reading.
+ */
+class OrcUtils {
+
+ private static final long MILLIS_PER_DAY = 86400000; // = 24 * 60 * 60 * 1000
+ private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
+ /**
+ * Converts an ORC schema to a Flink TypeInformation.
+ *
+ * @param schema The ORC schema.
+ * @return The TypeInformation that corresponds to the ORC schema.
+ */
+ static TypeInformation schemaToTypeInfo(TypeDescription schema) {
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ return BasicTypeInfo.BOOLEAN_TYPE_INFO;
+ case BYTE:
+ return BasicTypeInfo.BYTE_TYPE_INFO;
+ case SHORT:
+ return BasicTypeInfo.SHORT_TYPE_INFO;
+ case INT:
+ return BasicTypeInfo.INT_TYPE_INFO;
+ case LONG:
+ return BasicTypeInfo.LONG_TYPE_INFO;
+ case FLOAT:
+ return BasicTypeInfo.FLOAT_TYPE_INFO;
+ case DOUBLE:
+ return BasicTypeInfo.DOUBLE_TYPE_INFO;
+ case DECIMAL:
+ return BasicTypeInfo.BIG_DEC_TYPE_INFO;
+ case STRING:
+ case CHAR:
+ case VARCHAR:
+ return BasicTypeInfo.STRING_TYPE_INFO;
+ case DATE:
+ return SqlTimeTypeInfo.DATE;
+ case TIMESTAMP:
+ return SqlTimeTypeInfo.TIMESTAMP;
+ case BINARY:
+ return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO;
+ case STRUCT:
+ List<TypeDescription> fieldSchemas = schema.getChildren();
+ TypeInformation[] fieldTypes = new TypeInformation[fieldSchemas.size()];
+ for (int i = 0; i < fieldSchemas.size(); i++) {
+ fieldTypes[i] = schemaToTypeInfo(fieldSchemas.get(i));
+ }
+ String[] fieldNames = schema.getFieldNames().toArray(new String[]{});
+ return new RowTypeInfo(fieldTypes, fieldNames);
+ case LIST:
+ TypeDescription elementSchema = schema.getChildren().get(0);
+ TypeInformation<?> elementType = schemaToTypeInfo(elementSchema);
+ // arrays of primitive types are handled as object arrays to support null values
+ return ObjectArrayTypeInfo.getInfoFor(elementType);
+ case MAP:
+ TypeDescription keySchema = schema.getChildren().get(0);
+ TypeDescription valSchema = schema.getChildren().get(1);
+ TypeInformation<?> keyType = schemaToTypeInfo(keySchema);
+ TypeInformation<?> valType = schemaToTypeInfo(valSchema);
+ return new MapTypeInfo<>(keyType, valType);
+ case UNION:
+ throw new UnsupportedOperationException("UNION type is not supported yet.");
+ default:
+ throw new IllegalArgumentException("Unknown type " + schema);
+ }
+ }
+
+ /**
+ * Fills an ORC batch into an array of Row.
+ *
+ * @param rows The batch of rows need to be filled.
+ * @param schema The schema of the ORC data.
+ * @param batch The ORC data.
+ * @param selectedFields The list of selected ORC fields.
+ * @return The number of rows that were filled.
+ */
+ static int fillRows(Row[] rows, TypeDescription schema, VectorizedRowBatch batch, int[] selectedFields) {
+
+ int rowsToRead = Math.min((int) batch.count(), rows.length);
+
+ List<TypeDescription> fieldTypes = schema.getChildren();
+ // read each selected field
+ for (int rowIdx = 0; rowIdx < selectedFields.length; rowIdx++) {
+ int orcIdx = selectedFields[rowIdx];
+ readField(rows, rowIdx, fieldTypes.get(orcIdx), batch.cols[orcIdx], null, rowsToRead);
+ }
+ return rowsToRead;
+ }
+
+ /**
+ * Reads a vector of data into an array of objects.
+ *
+ * @param vals The array that needs to be filled.
+ * @param fieldIdx If the vals array is an array of Row, the index of the field that needs to be filled.
+ * Otherwise a -1 must be passed and the data is directly filled into the array.
+ * @param schema The schema of the vector to read.
+ * @param vector The vector to read.
+ * @param lengthVector If the vector is of type List or Map, the number of sub-elements to read for each field. Otherwise, it must be null.
+ * @param childCount The number of vector entries to read.
+ */
+ private static void readField(Object[] vals, int fieldIdx, TypeDescription schema, ColumnVector vector, long[] lengthVector, int childCount) {
+
+ // check the type of the vector to decide how to read it.
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ }
+ break;
+ case BYTE:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ }
+ break;
+ case SHORT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ }
+ break;
+ case INT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ }
+ break;
+ case LONG:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ }
+ break;
+ case FLOAT:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ }
+ break;
+ case DOUBLE:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ }
+ break;
+ case CHAR:
+ case VARCHAR:
+ case STRING:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DATE:
+ if (vector.noNulls) {
+ readNonNullLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ } else {
+ readLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case TIMESTAMP:
+ if (vector.noNulls) {
+ readNonNullTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ } else {
+ readTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case BINARY:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DECIMAL:
+ if (vector.noNulls) {
+ readNonNullDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ else {
+ readDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case STRUCT:
+ if (vector.noNulls) {
+ readNonNullStructColumn(vals, fieldIdx, (StructColumnVector) vector, schema, lengthVector, childCount);
+ } else {
+ readStructColumn(vals, fieldIdx, (StructColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case LIST:
+ if (vector.noNulls) {
+ readNonNullListColumn(vals, fieldIdx, (ListColumnVector) vector, schema, lengthVector, childCount);
+ }
+ else {
+ readListColumn(vals, fieldIdx, (ListColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case MAP:
+ if (vector.noNulls) {
+ readNonNullMapColumn(vals, fieldIdx, (MapColumnVector) vector, schema, lengthVector, childCount);
+ }
+ else {
+ readMapColumn(vals, fieldIdx, (MapColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case UNION:
+ throw new UnsupportedOperationException("UNION type not supported yet");
+ default:
+ throw new IllegalArgumentException("Unknown type " + schema);
+ }
+ }
+
+ private static <T> void readNonNullLongColumn(Object[] vals, int fieldIdx, LongColumnVector vector, long[] lengthVector, int childCount,
+ LongFunction<T> reader, IntFunction<T[]> array) {
+
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (vector.isRepeating) { // fill complete column with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ fillColumnWithRepeatingValue(vals, fieldIdx, repeatingValue, childCount);
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = reader.apply(vector.vector[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, reader.apply(vector.vector[i]));
+ }
+ }
+ }
+ } else { // in a list
+ T[] temp;
+ int offset = 0;
+ if (vector.isRepeating) { // fill complete list with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ Arrays.fill(temp, repeatingValue);
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = reader.apply(vector.vector[offset++]);
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+ }
+
+ private static <T> void readNonNullDoubleColumn(Object[] vals, int fieldIdx, DoubleColumnVector vector, long[] lengthVector, int childCount,
+ DoubleFunction<T> reader, IntFunction<T[]> array) {
+
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (vector.isRepeating) { // fill complete column with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ fillColumnWithRepeatingValue(vals, fieldIdx, repeatingValue, childCount);
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = reader.apply(vector.vector[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, reader.apply(vector.vector[i]));
+ }
+ }
+ }
+ } else { // in a list
+ T[] temp;
+ int offset = 0;
+ if (vector.isRepeating) { // fill complete list with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ Arrays.fill(temp, repeatingValue);
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = reader.apply(vector.vector[offset++]);
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+ }
+
+ private static void readNonNullBytesColumnAsString(Object[] vals, int fieldIdx, BytesColumnVector bytes, long[] lengthVector, int childCount) {
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (bytes.isRepeating) { // fill complete column with first value
+ String repeatingValue = new String(bytes.vector[0], bytes.start[0], bytes.length[0]);
--- End diff --
Good point! I found that ORC is using `UTF-8` internally.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152531212
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcTableSource.java ---
@@ -0,0 +1,429 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
+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.RowTypeInfo;
+import org.apache.flink.orc.OrcRowInputFormat.Predicate;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.expressions.Attribute;
+import org.apache.flink.table.expressions.BinaryComparison;
+import org.apache.flink.table.expressions.EqualTo;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.GreaterThan;
+import org.apache.flink.table.expressions.GreaterThanOrEqual;
+import org.apache.flink.table.expressions.IsFalse;
+import org.apache.flink.table.expressions.IsNotNull;
+import org.apache.flink.table.expressions.IsNull;
+import org.apache.flink.table.expressions.IsTrue;
+import org.apache.flink.table.expressions.LessThan;
+import org.apache.flink.table.expressions.LessThanOrEqual;
+import org.apache.flink.table.expressions.Literal;
+import org.apache.flink.table.expressions.Not;
+import org.apache.flink.table.expressions.NotEqualTo;
+import org.apache.flink.table.expressions.Or;
+import org.apache.flink.table.expressions.UnaryExpression;
+import org.apache.flink.table.sources.BatchTableSource;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.types.Row;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.orc.TypeDescription;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * A TableSource to read ORC files.
+ *
+ * <p>The {@link OrcTableSource} supports projection and filter push-down.</p>
+ *
+ * <p>An {@link OrcTableSource} is used as shown in the example below.
+ *
+ * <pre>
+ * {@code
+ * String path = "file:///my/data/file.orc";
+ * String schema = "struct<col1:boolean,col2:tinyint,col3:smallint,col4:int>"
+ * OrcTableSource orcSrc = new OrcTableSource(path, schema);
+ * tEnv.registerTableSource("orcTable", orcSrc);
+ * Table res = tableEnv.sql("SELECT * FROM orcTable");
+ * }
+ * </pre>
+ */
+public class OrcTableSource
+ implements BatchTableSource<Row>, ProjectableTableSource<Row>, FilterableTableSource<Row> {
+
+ private static final int DEFAULT_BATCH_SIZE = 1024;
+
+ // path to read ORC files from
+ private final String path;
+ // schema of the ORC file
+ private final TypeDescription orcSchema;
+ // the schema of the Table
+ private final TableSchema tableSchema;
+ // the configuration to read the file
+ private final Configuration orcConfig;
+ // the number of rows to read in a batch
+ private final int batchSize;
+
+ // type information of the data returned by the InputFormat
+ private final RowTypeInfo typeInfo;
+ // list of selected ORC fields to return
+ private final int[] selectedFields;
+ // list of predicates to apply
+ private final Predicate[] predicates;
+
+ /**
+ * Creates an OrcTableSouce with empty configuration.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as String.
+ */
+ public OrcTableSource(String path, String orcSchema) {
+ this(path, orcSchema, new Configuration(), DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcTableSource from an ORC schema string.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files.
+ */
+ public OrcTableSource(String path, String orcSchema, Configuration orcConfig) {
+ this(path, TypeDescription.fromString(orcSchema), orcConfig, DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcTableSource from an ORC schema string.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files.
+ * @param batchSize The number of Rows to read in a batch, default is 1000.
+ */
+ public OrcTableSource(String path, String orcSchema, Configuration orcConfig, int batchSize) {
+ this(path, TypeDescription.fromString(orcSchema), orcConfig, batchSize);
+ }
+
+ /**
+ * Creates an OrcTableSouce from an ORC TypeDescription.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as TypeDescription.
+ * @param orcConfig The configuration to read the ORC files.
+ * @param batchSize The number of Rows to read in a batch, default is 1000.
+ */
+ public OrcTableSource(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize) {
+ this(path, orcSchema, orcConfig, batchSize, null, null);
+ }
+
+ private OrcTableSource(String path, TypeDescription orcSchema, Configuration orcConfig,
+ int batchSize, int[] selectedFields, Predicate[] predicates) {
+
+ this.path = path;
+ this.orcSchema = orcSchema;
+ this.orcConfig = orcConfig;
+ this.batchSize = batchSize;
+ this.selectedFields = selectedFields;
+ this.predicates = predicates;
+
+ // determine the type information from the ORC schema
+ RowTypeInfo typeInfoFromSchema = (RowTypeInfo) OrcUtils.schemaToTypeInfo(this.orcSchema);
+
+ // set return type info
+ if (selectedFields == null) {
+ this.typeInfo = typeInfoFromSchema;
+ } else {
+ this.typeInfo = RowTypeInfo.projectFields(typeInfoFromSchema, selectedFields);
+ }
+
+ // create a TableSchema that corresponds to the ORC schema
+ this.tableSchema = new TableSchema(
+ typeInfoFromSchema.getFieldNames(),
+ typeInfoFromSchema.getFieldTypes()
+ );
+ }
+
+ @Override
+ public DataSet<Row> getDataSet(ExecutionEnvironment execEnv) {
+ OrcRowInputFormat orcIF = buildOrcInputFormat();
+ if (selectedFields != null) {
+ orcIF.selectFields(selectedFields);
+ }
+ if (predicates != null) {
+ for (OrcRowInputFormat.Predicate pred : predicates) {
+ orcIF.addPredicate(pred);
+ }
+ }
+ return execEnv.createInput(orcIF);
+ }
+
+ @VisibleForTesting
+ protected OrcRowInputFormat buildOrcInputFormat() {
+ return new OrcRowInputFormat(path, orcSchema, orcConfig, batchSize);
+ }
+
+ @Override
+ public TypeInformation<Row> getReturnType() {
+ return typeInfo;
+ }
+
+ @Override
+ public TableSchema getTableSchema() {
+ return this.tableSchema;
+ }
+
+ @Override
+ public TableSource<Row> projectFields(int[] selectedFields) {
+ // create a copy of the OrcTableSouce with new selected fields
+ return new OrcTableSource(path, orcSchema, orcConfig, batchSize, selectedFields, predicates);
+ }
+
+ @Override
+ public TableSource<Row> applyPredicate(List<Expression> predicates) {
+ ArrayList<Predicate> orcPredicates = new ArrayList<>();
+
+ // we do not remove any predicates from the list because ORC does not fully apply predicates
+ for (Expression pred : predicates) {
+ Predicate orcPred = toOrcPredicate(pred);
+ if (orcPred != null) {
+ orcPredicates.add(orcPred);
+ }
+ }
+
+ return new OrcTableSource(path, orcSchema, orcConfig, batchSize, selectedFields, orcPredicates.toArray(new Predicate[]{}));
+ }
+
+ @Override
+ public boolean isFilterPushedDown() {
+ return this.predicates != null;
+ }
+
+ @Override
+ public String explainSource() {
+ return "OrcFile[path=" + path + ", schema=" + orcSchema + ", filter=" + predicateString() + "]";
+ }
+
+ private String predicateString() {
+ if (predicates != null) {
+ return "AND(" + Arrays.toString(predicates) + ")";
+ } else {
+ return "TRUE";
+ }
+ }
+
+ // Predicate conversion for filter push-down.
+
+ private Predicate toOrcPredicate(Expression pred) {
+ if (pred instanceof Or) {
+ Predicate c1 = toOrcPredicate(((Or) pred).left());
+ Predicate c2 = toOrcPredicate(((Or) pred).right());
+ if (c1 == null || c2 == null) {
+ return null;
+ } else {
+ return new OrcRowInputFormat.Or(c1, c2);
+ }
+ } else if (pred instanceof Not) {
+ Predicate c = toOrcPredicate(((Not) pred).child());
+ if (c == null) {
+ return null;
+ } else {
+ return new OrcRowInputFormat.Not(c);
+ }
+ } else if (pred instanceof BinaryComparison) {
+
+ BinaryComparison binComp = (BinaryComparison) pred;
+
+ if (!isValid(binComp)) {
+ // not a valid predicate
+ return null;
+ }
+ PredicateLeaf.Type litType = getLiteralType(binComp);
+ if (litType == null) {
+ // unsupported literal type
+ return null;
+ }
+
+ boolean literalOnRight = literalOnRight(binComp);
+ String colName = getColumnName(binComp);
+ Serializable literal = (Serializable) getLiteral(binComp);
+
+ if (pred instanceof EqualTo) {
+ return new OrcRowInputFormat.Equals(colName, litType, literal);
+ } else if (pred instanceof NotEqualTo) {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.Equals(colName, litType, literal));
+ } else if (pred instanceof GreaterThan) {
+ if (literalOnRight) {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.LessThanEquals(colName, litType, literal));
+ } else {
+ return new OrcRowInputFormat.LessThan(colName, litType, literal);
+ }
+ } else if (pred instanceof GreaterThanOrEqual) {
+ if (literalOnRight) {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.LessThan(colName, litType, literal));
+ } else {
+ return new OrcRowInputFormat.LessThanEquals(colName, litType, literal);
+ }
+ } else if (pred instanceof LessThan) {
+ if (literalOnRight) {
+ return new OrcRowInputFormat.LessThan(colName, litType, literal);
+ } else {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.LessThanEquals(colName, litType, literal));
+ }
+ } else if (pred instanceof LessThanOrEqual) {
+ if (literalOnRight) {
+ return new OrcRowInputFormat.LessThanEquals(colName, litType, literal);
+ } else {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.LessThan(colName, litType, literal));
+ }
+ } else {
+ // unsupported predicate
+ return null;
+ }
+ } else if (pred instanceof UnaryExpression) {
+
+ UnaryExpression unary = (UnaryExpression) pred;
+ if (!isValid(unary)) {
+ // not a valid predicate
+ return null;
+ }
+ PredicateLeaf.Type colType = toOrcType(((UnaryExpression) pred).child().resultType());
+ if (colType == null) {
+ // unsupported type
+ return null;
+ }
+
+ String colName = getColumnName(unary);
+
+ if (pred instanceof IsNull) {
--- End diff --
I don't know how the ORC predicates are evaluated but if they comply with SQLs three-valued logic you cannot use this transformations. See: https://en.wikibooks.org/wiki/Structured_Query_Language/NULLs_and_the_Three_Valued_Logic#Step_2:_Boolean_Operations_within_3VL and `ScalarOperators`.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152513215
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcUtils.java ---
@@ -0,0 +1,1511 @@
+/*
+ * 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.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+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.types.Row;
+
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+
+import org.apache.orc.TypeDescription;
+
+import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.function.DoubleFunction;
+import java.util.function.IntFunction;
+import java.util.function.LongFunction;
+
+/**
+ * A class that provides utility methods for orc file reading.
+ */
+class OrcUtils {
+
+ private static final long MILLIS_PER_DAY = 86400000; // = 24 * 60 * 60 * 1000
+ private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
+ /**
+ * Converts an ORC schema to a Flink TypeInformation.
+ *
+ * @param schema The ORC schema.
+ * @return The TypeInformation that corresponds to the ORC schema.
+ */
+ static TypeInformation schemaToTypeInfo(TypeDescription schema) {
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ return BasicTypeInfo.BOOLEAN_TYPE_INFO;
+ case BYTE:
+ return BasicTypeInfo.BYTE_TYPE_INFO;
+ case SHORT:
+ return BasicTypeInfo.SHORT_TYPE_INFO;
+ case INT:
+ return BasicTypeInfo.INT_TYPE_INFO;
+ case LONG:
+ return BasicTypeInfo.LONG_TYPE_INFO;
+ case FLOAT:
+ return BasicTypeInfo.FLOAT_TYPE_INFO;
+ case DOUBLE:
+ return BasicTypeInfo.DOUBLE_TYPE_INFO;
+ case DECIMAL:
+ return BasicTypeInfo.BIG_DEC_TYPE_INFO;
+ case STRING:
+ case CHAR:
+ case VARCHAR:
+ return BasicTypeInfo.STRING_TYPE_INFO;
+ case DATE:
+ return SqlTimeTypeInfo.DATE;
+ case TIMESTAMP:
+ return SqlTimeTypeInfo.TIMESTAMP;
+ case BINARY:
+ return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO;
+ case STRUCT:
+ List<TypeDescription> fieldSchemas = schema.getChildren();
+ TypeInformation[] fieldTypes = new TypeInformation[fieldSchemas.size()];
+ for (int i = 0; i < fieldSchemas.size(); i++) {
+ fieldTypes[i] = schemaToTypeInfo(fieldSchemas.get(i));
+ }
+ String[] fieldNames = schema.getFieldNames().toArray(new String[]{});
+ return new RowTypeInfo(fieldTypes, fieldNames);
+ case LIST:
+ TypeDescription elementSchema = schema.getChildren().get(0);
+ TypeInformation<?> elementType = schemaToTypeInfo(elementSchema);
+ // arrays of primitive types are handled as object arrays to support null values
+ return ObjectArrayTypeInfo.getInfoFor(elementType);
+ case MAP:
+ TypeDescription keySchema = schema.getChildren().get(0);
+ TypeDescription valSchema = schema.getChildren().get(1);
+ TypeInformation<?> keyType = schemaToTypeInfo(keySchema);
+ TypeInformation<?> valType = schemaToTypeInfo(valSchema);
+ return new MapTypeInfo<>(keyType, valType);
+ case UNION:
+ throw new UnsupportedOperationException("UNION type is not supported yet.");
+ default:
+ throw new IllegalArgumentException("Unknown type " + schema);
+ }
+ }
+
+ /**
+ * Fills an ORC batch into an array of Row.
+ *
+ * @param rows The batch of rows need to be filled.
+ * @param schema The schema of the ORC data.
+ * @param batch The ORC data.
+ * @param selectedFields The list of selected ORC fields.
+ * @return The number of rows that were filled.
+ */
+ static int fillRows(Row[] rows, TypeDescription schema, VectorizedRowBatch batch, int[] selectedFields) {
+
+ int rowsToRead = Math.min((int) batch.count(), rows.length);
+
+ List<TypeDescription> fieldTypes = schema.getChildren();
+ // read each selected field
+ for (int rowIdx = 0; rowIdx < selectedFields.length; rowIdx++) {
+ int orcIdx = selectedFields[rowIdx];
+ readField(rows, rowIdx, fieldTypes.get(orcIdx), batch.cols[orcIdx], null, rowsToRead);
+ }
+ return rowsToRead;
+ }
+
+ /**
+ * Reads a vector of data into an array of objects.
+ *
+ * @param vals The array that needs to be filled.
+ * @param fieldIdx If the vals array is an array of Row, the index of the field that needs to be filled.
+ * Otherwise a -1 must be passed and the data is directly filled into the array.
+ * @param schema The schema of the vector to read.
+ * @param vector The vector to read.
+ * @param lengthVector If the vector is of type List or Map, the number of sub-elements to read for each field. Otherwise, it must be null.
+ * @param childCount The number of vector entries to read.
+ */
+ private static void readField(Object[] vals, int fieldIdx, TypeDescription schema, ColumnVector vector, long[] lengthVector, int childCount) {
+
+ // check the type of the vector to decide how to read it.
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ }
+ break;
+ case BYTE:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ }
+ break;
+ case SHORT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ }
+ break;
+ case INT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ }
+ break;
+ case LONG:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ }
+ break;
+ case FLOAT:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ }
+ break;
+ case DOUBLE:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ }
+ break;
+ case CHAR:
+ case VARCHAR:
+ case STRING:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DATE:
+ if (vector.noNulls) {
+ readNonNullLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ } else {
+ readLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case TIMESTAMP:
+ if (vector.noNulls) {
+ readNonNullTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ } else {
+ readTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case BINARY:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DECIMAL:
+ if (vector.noNulls) {
+ readNonNullDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ else {
+ readDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case STRUCT:
+ if (vector.noNulls) {
+ readNonNullStructColumn(vals, fieldIdx, (StructColumnVector) vector, schema, lengthVector, childCount);
+ } else {
+ readStructColumn(vals, fieldIdx, (StructColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case LIST:
+ if (vector.noNulls) {
+ readNonNullListColumn(vals, fieldIdx, (ListColumnVector) vector, schema, lengthVector, childCount);
+ }
+ else {
+ readListColumn(vals, fieldIdx, (ListColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case MAP:
+ if (vector.noNulls) {
+ readNonNullMapColumn(vals, fieldIdx, (MapColumnVector) vector, schema, lengthVector, childCount);
+ }
+ else {
+ readMapColumn(vals, fieldIdx, (MapColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case UNION:
+ throw new UnsupportedOperationException("UNION type not supported yet");
+ default:
+ throw new IllegalArgumentException("Unknown type " + schema);
+ }
+ }
+
+ private static <T> void readNonNullLongColumn(Object[] vals, int fieldIdx, LongColumnVector vector, long[] lengthVector, int childCount,
+ LongFunction<T> reader, IntFunction<T[]> array) {
+
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (vector.isRepeating) { // fill complete column with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ fillColumnWithRepeatingValue(vals, fieldIdx, repeatingValue, childCount);
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = reader.apply(vector.vector[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, reader.apply(vector.vector[i]));
+ }
+ }
+ }
+ } else { // in a list
+ T[] temp;
+ int offset = 0;
+ if (vector.isRepeating) { // fill complete list with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ Arrays.fill(temp, repeatingValue);
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = reader.apply(vector.vector[offset++]);
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+ }
+
+ private static <T> void readNonNullDoubleColumn(Object[] vals, int fieldIdx, DoubleColumnVector vector, long[] lengthVector, int childCount,
+ DoubleFunction<T> reader, IntFunction<T[]> array) {
+
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (vector.isRepeating) { // fill complete column with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ fillColumnWithRepeatingValue(vals, fieldIdx, repeatingValue, childCount);
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = reader.apply(vector.vector[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, reader.apply(vector.vector[i]));
+ }
+ }
+ }
+ } else { // in a list
+ T[] temp;
+ int offset = 0;
+ if (vector.isRepeating) { // fill complete list with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ Arrays.fill(temp, repeatingValue);
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = reader.apply(vector.vector[offset++]);
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+ }
+
+ private static void readNonNullBytesColumnAsString(Object[] vals, int fieldIdx, BytesColumnVector bytes, long[] lengthVector, int childCount) {
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (bytes.isRepeating) { // fill complete column with first value
+ String repeatingValue = new String(bytes.vector[0], bytes.start[0], bytes.length[0]);
+ fillColumnWithRepeatingValue(vals, fieldIdx, repeatingValue, childCount);
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = new String(bytes.vector[i], bytes.start[i], bytes.length[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, new String(bytes.vector[i], bytes.start[i], bytes.length[i]));
+ }
+ }
+ }
+ } else {
+ String[] temp;
+ int offset = 0;
+ if (bytes.isRepeating) { // fill complete list with first value
+ String repeatingValue = new String(bytes.vector[0], bytes.start[0], bytes.length[0]);
+ for (int i = 0; offset < childCount; i++) {
+ temp = new String[(int) lengthVector[i]];
+ Arrays.fill(temp, repeatingValue);
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = new String[(int) lengthVector[i]];
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = new String(bytes.vector[offset], bytes.start[offset], bytes.length[offset]);
+ offset++;
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+ }
+
+ private static void readNonNullBytesColumnAsBinary(Object[] vals, int fieldIdx, BytesColumnVector bytes, long[] lengthVector, int childCount) {
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (bytes.isRepeating) { // fill complete column with first value
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ // don't reuse repeating val to avoid object mutation
+ vals[i] = readBinary(bytes.vector[0], bytes.start[0], bytes.length[0]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ // don't reuse repeating val to avoid object mutation
+ rows[i].setField(fieldIdx, readBinary(bytes.vector[0], bytes.start[0], bytes.length[0]));
+ }
+ }
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = readBinary(bytes.vector[i], bytes.start[i], bytes.length[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, readBinary(bytes.vector[i], bytes.start[i], bytes.length[i]));
+ }
+ }
+ }
+ } else {
+ byte[][] temp;
+ int offset = 0;
+ if (bytes.isRepeating) { // fill complete list with first value
+ for (int i = 0; offset < childCount; i++) {
+ temp = new byte[(int) lengthVector[i]][];
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = readBinary(bytes.vector[0], bytes.start[0], bytes.length[0]);
+ }
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = new byte[(int) lengthVector[i]][];
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = readBinary(bytes.vector[offset], bytes.start[offset], bytes.length[offset]);
+ offset++;
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+ }
+
+ private static void readNonNullLongColumnAsDate(Object[] vals, int fieldIdx, LongColumnVector vector, long[] lengthVector, int childCount) {
+
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (vector.isRepeating) { // fill complete column with first value
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ // do not reuse repeated value due to mutability of Date
+ vals[i] = readDate(vector.vector[0]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ // do not reuse repeated value due to mutability of Date
+ rows[i].setField(fieldIdx, readDate(vector.vector[0]));
+ }
+ }
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = readDate(vector.vector[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, readDate(vector.vector[i]));
+ }
+ }
+ }
+ } else { // in a list
+ Date[] temp;
+ int offset = 0;
+ if (vector.isRepeating) { // fill complete list with first value
+ for (int i = 0; offset < childCount; i++) {
+ temp = new Date[(int) lengthVector[i]];
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = readDate(vector.vector[0]);
+ }
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = new Date[(int) lengthVector[i]];
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = readDate(vector.vector[offset++]);
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+ }
+
+ private static void readNonNullTimestampColumn(Object[] vals, int fieldIdx, TimestampColumnVector vector, long[] lengthVector, int childCount) {
+
+ // check if the timestamps need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (vector.isRepeating) { // fill complete column with first value
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ // do not reuse value to prevent object mutation
+ vals[i] = readTimestamp(vector.time[0], vector.nanos[0]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ // do not reuse value to prevent object mutation
+ rows[i].setField(fieldIdx, readTimestamp(vector.time[0], vector.nanos[0]));
+ }
+ }
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = readTimestamp(vector.time[i], vector.nanos[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, readTimestamp(vector.time[i], vector.nanos[i]));
+ }
+ }
+ }
+ } else {
+ Timestamp[] temp;
+ int offset = 0;
+ if (vector.isRepeating) { // fill complete list with first value
+ for (int i = 0; offset < childCount; i++) {
+ temp = new Timestamp[(int) lengthVector[i]];
+ for (int j = 0; j < temp.length; j++) {
+ // do not reuse value to prevent object mutation
+ temp[j] = readTimestamp(vector.time[0], vector.nanos[0]);
+ }
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = new Timestamp[(int) lengthVector[i]];
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = readTimestamp(vector.time[offset], vector.nanos[offset]);
+ offset++;
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+ }
+
+ private static void readNonNullDecimalColumn(Object[] vals, int fieldIdx, DecimalColumnVector vector, long[] lengthVector, int childCount) {
+
+ // check if the decimals need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (vector.isRepeating) { // fill complete column with first value
+ fillColumnWithRepeatingValue(vals, fieldIdx, readBigDecimal(vector.vector[0]), childCount);
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = readBigDecimal(vector.vector[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, readBigDecimal(vector.vector[i]));
+ }
+ }
+ }
+ } else {
+ BigDecimal[] temp;
+ int offset = 0;
+ if (vector.isRepeating) { // fill complete list with first value
+ BigDecimal repeatingValue = readBigDecimal(vector.vector[0]);
+ for (int i = 0; offset < childCount; i++) {
+ temp = new BigDecimal[(int) lengthVector[i]];
+ Arrays.fill(temp, repeatingValue);
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = new BigDecimal[(int) lengthVector[i]];
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = readBigDecimal(vector.vector[offset++]);
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+
+ }
+
+ private static void readNonNullStructColumn(Object[] vals, int fieldIdx, StructColumnVector structVector, TypeDescription schema, long[] lengthVector, int childCount) {
+
+ List<TypeDescription> childrenTypes = schema.getChildren();
+
+ int numFields = childrenTypes.size();
+ // create a batch of Rows to read the structs
+ Row[] structs = new Row[childCount];
+ // TODO: possible improvement: reuse existing Row objects
+ for (int i = 0; i < childCount; i++) {
+ structs[i] = new Row(numFields);
+ }
+
+ // read struct fields
+ for (int i = 0; i < numFields; i++) {
+ readField(structs, i, childrenTypes.get(i), structVector.fields[i], null, childCount);
+ }
+
+ // check if the structs need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (fieldIdx == -1) { // set struct as an object
+ System.arraycopy(structs, 0, vals, 0, childCount);
+ } else { // set struct as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, structs[i]);
+ }
+ }
+ } else { // struct in a list
+ int offset = 0;
+ Row[] temp;
+ for (int i = 0; offset < childCount; i++) {
+ temp = new Row[(int) lengthVector[i]];
+ System.arraycopy(structs, offset, temp, 0, temp.length);
+ offset = offset + temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+
+ private static void readNonNullListColumn(Object[] vals, int fieldIdx, ListColumnVector list, TypeDescription schema, long[] lengthVector, int childCount) {
+
+ TypeDescription fieldType = schema.getChildren().get(0);
+ // check if the list need to be read into lists or as single values
+ if (lengthVector == null) {
+ long[] lengthVectorNested = list.lengths;
+ readField(vals, fieldIdx, fieldType, list.child, lengthVectorNested, list.childCount);
+ } else { // list in a list
+ Object[] nestedLists = new Object[childCount];
+ // length vector for nested list
+ long[] lengthVectorNested = list.lengths;
+ // read nested list
+ readField(nestedLists, -1, fieldType, list.child, lengthVectorNested, list.childCount);
+ // get type of nestedList
+ Class<?> classType = nestedLists[0].getClass();
+
+ // fill outer list with nested list
+ int offset = 0;
+ int length;
+ for (int i = 0; offset < childCount; i++) {
+ length = (int) lengthVector[i];
+ Object[] temp = (Object[]) Array.newInstance(classType, length);
+ System.arraycopy(nestedLists, offset, temp, 0, length);
+ offset = offset + length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+
+ private static void readNonNullMapColumn(Object[] vals, int fieldIdx, MapColumnVector mapsVector, TypeDescription schema, long[] lengthVector, int childCount) {
+
+ List<TypeDescription> fieldType = schema.getChildren();
+ TypeDescription keyType = fieldType.get(0);
+ TypeDescription valueType = fieldType.get(1);
+
+ ColumnVector keys = mapsVector.keys;
+ ColumnVector values = mapsVector.values;
+ Object[] keyRows = new Object[mapsVector.childCount];
+ Object[] valueRows = new Object[mapsVector.childCount];
+
+ // read map kes and values
--- End diff --
keys
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/5043
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152510004
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcUtils.java ---
@@ -0,0 +1,1511 @@
+/*
+ * 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.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+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.types.Row;
+
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+
+import org.apache.orc.TypeDescription;
+
+import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.function.DoubleFunction;
+import java.util.function.IntFunction;
+import java.util.function.LongFunction;
+
+/**
+ * A class that provides utility methods for orc file reading.
+ */
+class OrcUtils {
+
+ private static final long MILLIS_PER_DAY = 86400000; // = 24 * 60 * 60 * 1000
+ private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
+ /**
+ * Converts an ORC schema to a Flink TypeInformation.
+ *
+ * @param schema The ORC schema.
+ * @return The TypeInformation that corresponds to the ORC schema.
+ */
+ static TypeInformation schemaToTypeInfo(TypeDescription schema) {
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ return BasicTypeInfo.BOOLEAN_TYPE_INFO;
+ case BYTE:
+ return BasicTypeInfo.BYTE_TYPE_INFO;
+ case SHORT:
+ return BasicTypeInfo.SHORT_TYPE_INFO;
+ case INT:
+ return BasicTypeInfo.INT_TYPE_INFO;
+ case LONG:
+ return BasicTypeInfo.LONG_TYPE_INFO;
+ case FLOAT:
+ return BasicTypeInfo.FLOAT_TYPE_INFO;
+ case DOUBLE:
+ return BasicTypeInfo.DOUBLE_TYPE_INFO;
+ case DECIMAL:
+ return BasicTypeInfo.BIG_DEC_TYPE_INFO;
+ case STRING:
+ case CHAR:
+ case VARCHAR:
+ return BasicTypeInfo.STRING_TYPE_INFO;
+ case DATE:
+ return SqlTimeTypeInfo.DATE;
+ case TIMESTAMP:
+ return SqlTimeTypeInfo.TIMESTAMP;
+ case BINARY:
+ return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO;
+ case STRUCT:
+ List<TypeDescription> fieldSchemas = schema.getChildren();
+ TypeInformation[] fieldTypes = new TypeInformation[fieldSchemas.size()];
+ for (int i = 0; i < fieldSchemas.size(); i++) {
+ fieldTypes[i] = schemaToTypeInfo(fieldSchemas.get(i));
+ }
+ String[] fieldNames = schema.getFieldNames().toArray(new String[]{});
+ return new RowTypeInfo(fieldTypes, fieldNames);
+ case LIST:
+ TypeDescription elementSchema = schema.getChildren().get(0);
+ TypeInformation<?> elementType = schemaToTypeInfo(elementSchema);
+ // arrays of primitive types are handled as object arrays to support null values
+ return ObjectArrayTypeInfo.getInfoFor(elementType);
+ case MAP:
+ TypeDescription keySchema = schema.getChildren().get(0);
+ TypeDescription valSchema = schema.getChildren().get(1);
+ TypeInformation<?> keyType = schemaToTypeInfo(keySchema);
+ TypeInformation<?> valType = schemaToTypeInfo(valSchema);
+ return new MapTypeInfo<>(keyType, valType);
+ case UNION:
+ throw new UnsupportedOperationException("UNION type is not supported yet.");
+ default:
+ throw new IllegalArgumentException("Unknown type " + schema);
+ }
+ }
+
+ /**
+ * Fills an ORC batch into an array of Row.
+ *
+ * @param rows The batch of rows need to be filled.
+ * @param schema The schema of the ORC data.
+ * @param batch The ORC data.
+ * @param selectedFields The list of selected ORC fields.
+ * @return The number of rows that were filled.
+ */
+ static int fillRows(Row[] rows, TypeDescription schema, VectorizedRowBatch batch, int[] selectedFields) {
+
+ int rowsToRead = Math.min((int) batch.count(), rows.length);
+
+ List<TypeDescription> fieldTypes = schema.getChildren();
+ // read each selected field
+ for (int rowIdx = 0; rowIdx < selectedFields.length; rowIdx++) {
+ int orcIdx = selectedFields[rowIdx];
+ readField(rows, rowIdx, fieldTypes.get(orcIdx), batch.cols[orcIdx], null, rowsToRead);
+ }
+ return rowsToRead;
+ }
+
+ /**
+ * Reads a vector of data into an array of objects.
+ *
+ * @param vals The array that needs to be filled.
+ * @param fieldIdx If the vals array is an array of Row, the index of the field that needs to be filled.
+ * Otherwise a -1 must be passed and the data is directly filled into the array.
+ * @param schema The schema of the vector to read.
+ * @param vector The vector to read.
+ * @param lengthVector If the vector is of type List or Map, the number of sub-elements to read for each field. Otherwise, it must be null.
+ * @param childCount The number of vector entries to read.
+ */
+ private static void readField(Object[] vals, int fieldIdx, TypeDescription schema, ColumnVector vector, long[] lengthVector, int childCount) {
+
+ // check the type of the vector to decide how to read it.
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ }
+ break;
+ case BYTE:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ }
+ break;
+ case SHORT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ }
+ break;
+ case INT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ }
+ break;
+ case LONG:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ }
+ break;
+ case FLOAT:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ }
+ break;
+ case DOUBLE:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ }
+ break;
+ case CHAR:
+ case VARCHAR:
+ case STRING:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DATE:
+ if (vector.noNulls) {
+ readNonNullLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ } else {
+ readLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case TIMESTAMP:
+ if (vector.noNulls) {
+ readNonNullTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ } else {
+ readTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case BINARY:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DECIMAL:
+ if (vector.noNulls) {
+ readNonNullDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ else {
+ readDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case STRUCT:
+ if (vector.noNulls) {
+ readNonNullStructColumn(vals, fieldIdx, (StructColumnVector) vector, schema, lengthVector, childCount);
+ } else {
+ readStructColumn(vals, fieldIdx, (StructColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case LIST:
+ if (vector.noNulls) {
+ readNonNullListColumn(vals, fieldIdx, (ListColumnVector) vector, schema, lengthVector, childCount);
+ }
+ else {
+ readListColumn(vals, fieldIdx, (ListColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case MAP:
+ if (vector.noNulls) {
+ readNonNullMapColumn(vals, fieldIdx, (MapColumnVector) vector, schema, lengthVector, childCount);
+ }
+ else {
+ readMapColumn(vals, fieldIdx, (MapColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case UNION:
+ throw new UnsupportedOperationException("UNION type not supported yet");
+ default:
+ throw new IllegalArgumentException("Unknown type " + schema);
+ }
+ }
+
+ private static <T> void readNonNullLongColumn(Object[] vals, int fieldIdx, LongColumnVector vector, long[] lengthVector, int childCount,
+ LongFunction<T> reader, IntFunction<T[]> array) {
+
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (vector.isRepeating) { // fill complete column with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ fillColumnWithRepeatingValue(vals, fieldIdx, repeatingValue, childCount);
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = reader.apply(vector.vector[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, reader.apply(vector.vector[i]));
+ }
+ }
+ }
+ } else { // in a list
+ T[] temp;
+ int offset = 0;
+ if (vector.isRepeating) { // fill complete list with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ Arrays.fill(temp, repeatingValue);
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = reader.apply(vector.vector[offset++]);
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+ }
+
+ private static <T> void readNonNullDoubleColumn(Object[] vals, int fieldIdx, DoubleColumnVector vector, long[] lengthVector, int childCount,
+ DoubleFunction<T> reader, IntFunction<T[]> array) {
+
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (vector.isRepeating) { // fill complete column with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ fillColumnWithRepeatingValue(vals, fieldIdx, repeatingValue, childCount);
+ } else {
+ if (fieldIdx == -1) { // set as an object
+ for (int i = 0; i < childCount; i++) {
+ vals[i] = reader.apply(vector.vector[i]);
+ }
+ } else { // set as a field of Row
+ Row[] rows = (Row[]) vals;
+ for (int i = 0; i < childCount; i++) {
+ rows[i].setField(fieldIdx, reader.apply(vector.vector[i]));
+ }
+ }
+ }
+ } else { // in a list
+ T[] temp;
+ int offset = 0;
+ if (vector.isRepeating) { // fill complete list with first value
+ T repeatingValue = reader.apply(vector.vector[0]);
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ Arrays.fill(temp, repeatingValue);
+ offset += temp.length;
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ } else {
+ for (int i = 0; offset < childCount; i++) {
+ temp = array.apply((int) lengthVector[i]);
+ for (int j = 0; j < temp.length; j++) {
+ temp[j] = reader.apply(vector.vector[offset++]);
+ }
+ if (fieldIdx == -1) {
+ vals[i] = temp;
+ } else {
+ ((Row) vals[i]).setField(fieldIdx, temp);
+ }
+ }
+ }
+ }
+ }
+
+ private static void readNonNullBytesColumnAsString(Object[] vals, int fieldIdx, BytesColumnVector bytes, long[] lengthVector, int childCount) {
+ // check if the values need to be read into lists or as single values
+ if (lengthVector == null) {
+ if (bytes.isRepeating) { // fill complete column with first value
+ String repeatingValue = new String(bytes.vector[0], bytes.start[0], bytes.length[0]);
--- End diff --
Do we need to consider the Charset here?
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152578874
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcRowInputFormat.java ---
@@ -0,0 +1,747 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.StripeInformation;
+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 java.io.Serializable;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.orc.OrcUtils.fillRows;
+
+/**
+ * InputFormat to read ORC files.
+ */
+public class OrcRowInputFormat extends FileInputFormat<Row> implements ResultTypeQueryable<Row> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrcRowInputFormat.class);
+ // the number of rows read in a batch
+ private static final int DEFAULT_BATCH_SIZE = 1000;
+
+ // the number of fields rows to read in a batch
+ private int batchSize;
+ // the configuration to read with
+ private Configuration conf;
--- End diff --
Ok, sorry, I didn't check for custom serialization.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152571674
--- Diff: flink-core/src/main/java/org/apache/flink/api/java/typeutils/RowTypeInfo.java ---
@@ -355,4 +355,21 @@ public void addComparatorField(int fieldId, TypeComparator<?> comparator) {
comparatorOrders);
}
}
+
+ /**
+ * Creates a {@link RowTypeInfo} with projected fields.
+ *
+ * @param rowType The original RowTypeInfo whose fields are projected
+ * @param fieldMapping The field mapping of the projection
+ * @return A RowTypeInfo with projected fields.
+ */
+ public static RowTypeInfo projectFields(RowTypeInfo rowType, int[] fieldMapping) {
--- End diff --
I think it is cleaner to have this as a static method than an instance method.
A static method makes it explicit that this creates a new (immutable) `RowTypeInfo`.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152506343
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcUtils.java ---
@@ -0,0 +1,1511 @@
+/*
+ * 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.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+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.types.Row;
+
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+
+import org.apache.orc.TypeDescription;
+
+import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.function.DoubleFunction;
+import java.util.function.IntFunction;
+import java.util.function.LongFunction;
+
+/**
+ * A class that provides utility methods for orc file reading.
+ */
+class OrcUtils {
+
+ private static final long MILLIS_PER_DAY = 86400000; // = 24 * 60 * 60 * 1000
+ private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
+ /**
+ * Converts an ORC schema to a Flink TypeInformation.
+ *
+ * @param schema The ORC schema.
+ * @return The TypeInformation that corresponds to the ORC schema.
+ */
+ static TypeInformation schemaToTypeInfo(TypeDescription schema) {
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ return BasicTypeInfo.BOOLEAN_TYPE_INFO;
+ case BYTE:
+ return BasicTypeInfo.BYTE_TYPE_INFO;
+ case SHORT:
+ return BasicTypeInfo.SHORT_TYPE_INFO;
+ case INT:
+ return BasicTypeInfo.INT_TYPE_INFO;
+ case LONG:
+ return BasicTypeInfo.LONG_TYPE_INFO;
+ case FLOAT:
+ return BasicTypeInfo.FLOAT_TYPE_INFO;
+ case DOUBLE:
+ return BasicTypeInfo.DOUBLE_TYPE_INFO;
+ case DECIMAL:
+ return BasicTypeInfo.BIG_DEC_TYPE_INFO;
+ case STRING:
+ case CHAR:
+ case VARCHAR:
+ return BasicTypeInfo.STRING_TYPE_INFO;
+ case DATE:
+ return SqlTimeTypeInfo.DATE;
+ case TIMESTAMP:
+ return SqlTimeTypeInfo.TIMESTAMP;
+ case BINARY:
+ return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO;
+ case STRUCT:
+ List<TypeDescription> fieldSchemas = schema.getChildren();
+ TypeInformation[] fieldTypes = new TypeInformation[fieldSchemas.size()];
+ for (int i = 0; i < fieldSchemas.size(); i++) {
+ fieldTypes[i] = schemaToTypeInfo(fieldSchemas.get(i));
+ }
+ String[] fieldNames = schema.getFieldNames().toArray(new String[]{});
+ return new RowTypeInfo(fieldTypes, fieldNames);
+ case LIST:
+ TypeDescription elementSchema = schema.getChildren().get(0);
+ TypeInformation<?> elementType = schemaToTypeInfo(elementSchema);
+ // arrays of primitive types are handled as object arrays to support null values
+ return ObjectArrayTypeInfo.getInfoFor(elementType);
+ case MAP:
+ TypeDescription keySchema = schema.getChildren().get(0);
+ TypeDescription valSchema = schema.getChildren().get(1);
+ TypeInformation<?> keyType = schemaToTypeInfo(keySchema);
+ TypeInformation<?> valType = schemaToTypeInfo(valSchema);
+ return new MapTypeInfo<>(keyType, valType);
+ case UNION:
+ throw new UnsupportedOperationException("UNION type is not supported yet.");
+ default:
+ throw new IllegalArgumentException("Unknown type " + schema);
+ }
+ }
+
+ /**
+ * Fills an ORC batch into an array of Row.
+ *
+ * @param rows The batch of rows need to be filled.
+ * @param schema The schema of the ORC data.
+ * @param batch The ORC data.
+ * @param selectedFields The list of selected ORC fields.
+ * @return The number of rows that were filled.
+ */
+ static int fillRows(Row[] rows, TypeDescription schema, VectorizedRowBatch batch, int[] selectedFields) {
+
+ int rowsToRead = Math.min((int) batch.count(), rows.length);
+
+ List<TypeDescription> fieldTypes = schema.getChildren();
+ // read each selected field
+ for (int rowIdx = 0; rowIdx < selectedFields.length; rowIdx++) {
+ int orcIdx = selectedFields[rowIdx];
+ readField(rows, rowIdx, fieldTypes.get(orcIdx), batch.cols[orcIdx], null, rowsToRead);
+ }
+ return rowsToRead;
+ }
+
+ /**
+ * Reads a vector of data into an array of objects.
+ *
+ * @param vals The array that needs to be filled.
+ * @param fieldIdx If the vals array is an array of Row, the index of the field that needs to be filled.
+ * Otherwise a -1 must be passed and the data is directly filled into the array.
+ * @param schema The schema of the vector to read.
+ * @param vector The vector to read.
+ * @param lengthVector If the vector is of type List or Map, the number of sub-elements to read for each field. Otherwise, it must be null.
+ * @param childCount The number of vector entries to read.
+ */
+ private static void readField(Object[] vals, int fieldIdx, TypeDescription schema, ColumnVector vector, long[] lengthVector, int childCount) {
+
+ // check the type of the vector to decide how to read it.
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ }
+ break;
+ case BYTE:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ }
+ break;
+ case SHORT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ }
+ break;
+ case INT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ }
+ break;
+ case LONG:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ }
+ break;
+ case FLOAT:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ }
+ break;
+ case DOUBLE:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ }
+ break;
+ case CHAR:
+ case VARCHAR:
+ case STRING:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DATE:
+ if (vector.noNulls) {
+ readNonNullLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ } else {
+ readLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case TIMESTAMP:
+ if (vector.noNulls) {
+ readNonNullTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ } else {
+ readTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case BINARY:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DECIMAL:
+ if (vector.noNulls) {
+ readNonNullDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ else {
--- End diff --
Use a consistent formatting.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152531937
--- Diff: flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceITCase.java ---
@@ -0,0 +1,112 @@
+/*
+ * 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.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.BatchTableEnvironment;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for {@link OrcTableSource}.
+ */
+public class OrcTableSourceITCase extends MultipleProgramsTestBase {
--- End diff --
`TableProgramsTestBase` ?
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152577385
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalTableSourceScan.scala ---
@@ -62,7 +62,19 @@ class FlinkLogicalTableSourceScan(
override def computeSelfCost(planner: RelOptPlanner, metadata: RelMetadataQuery): RelOptCost = {
val rowCnt = metadata.getRowCount(this)
- planner.getCostFactory.makeCost(rowCnt, rowCnt, rowCnt * estimateRowSize(getRowType))
+
+ val adjustedCnt: Double = tableSource match {
+ case f: FilterableTableSource[_] if f.isFilterPushedDown =>
+ // ensure we prefer FilterableTableSources with pushed-down filters.
+ rowCnt - 1.0
--- End diff --
Doesn't really make a difference IMO. It's all about relative costs. We only need to make sure that a `FilterableTableSource` with pushed down predicates appears to be less expensive than the same `TableSource` without pushed predicates.
The problem has not occurred before, because the `OrcTableSource` does not guarantee that all emitted rows match the pushed predicates. Therefore, all predicates are also applied by a following `Calc` operator such that the cost of that operator is not decreased.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152533796
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcTableSource.java ---
@@ -0,0 +1,429 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
+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.RowTypeInfo;
+import org.apache.flink.orc.OrcRowInputFormat.Predicate;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.expressions.Attribute;
+import org.apache.flink.table.expressions.BinaryComparison;
+import org.apache.flink.table.expressions.EqualTo;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.GreaterThan;
+import org.apache.flink.table.expressions.GreaterThanOrEqual;
+import org.apache.flink.table.expressions.IsFalse;
+import org.apache.flink.table.expressions.IsNotNull;
+import org.apache.flink.table.expressions.IsNull;
+import org.apache.flink.table.expressions.IsTrue;
+import org.apache.flink.table.expressions.LessThan;
+import org.apache.flink.table.expressions.LessThanOrEqual;
+import org.apache.flink.table.expressions.Literal;
+import org.apache.flink.table.expressions.Not;
+import org.apache.flink.table.expressions.NotEqualTo;
+import org.apache.flink.table.expressions.Or;
+import org.apache.flink.table.expressions.UnaryExpression;
+import org.apache.flink.table.sources.BatchTableSource;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.types.Row;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.orc.TypeDescription;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * A TableSource to read ORC files.
+ *
+ * <p>The {@link OrcTableSource} supports projection and filter push-down.</p>
+ *
+ * <p>An {@link OrcTableSource} is used as shown in the example below.
+ *
+ * <pre>
+ * {@code
+ * String path = "file:///my/data/file.orc";
+ * String schema = "struct<col1:boolean,col2:tinyint,col3:smallint,col4:int>"
+ * OrcTableSource orcSrc = new OrcTableSource(path, schema);
+ * tEnv.registerTableSource("orcTable", orcSrc);
+ * Table res = tableEnv.sql("SELECT * FROM orcTable");
+ * }
+ * </pre>
+ */
+public class OrcTableSource
+ implements BatchTableSource<Row>, ProjectableTableSource<Row>, FilterableTableSource<Row> {
+
+ private static final int DEFAULT_BATCH_SIZE = 1024;
+
+ // path to read ORC files from
+ private final String path;
+ // schema of the ORC file
+ private final TypeDescription orcSchema;
+ // the schema of the Table
+ private final TableSchema tableSchema;
+ // the configuration to read the file
+ private final Configuration orcConfig;
+ // the number of rows to read in a batch
+ private final int batchSize;
+
+ // type information of the data returned by the InputFormat
+ private final RowTypeInfo typeInfo;
+ // list of selected ORC fields to return
+ private final int[] selectedFields;
+ // list of predicates to apply
+ private final Predicate[] predicates;
+
+ /**
+ * Creates an OrcTableSouce with empty configuration.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as String.
+ */
+ public OrcTableSource(String path, String orcSchema) {
+ this(path, orcSchema, new Configuration(), DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcTableSource from an ORC schema string.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files.
+ */
+ public OrcTableSource(String path, String orcSchema, Configuration orcConfig) {
+ this(path, TypeDescription.fromString(orcSchema), orcConfig, DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcTableSource from an ORC schema string.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files.
+ * @param batchSize The number of Rows to read in a batch, default is 1000.
+ */
+ public OrcTableSource(String path, String orcSchema, Configuration orcConfig, int batchSize) {
+ this(path, TypeDescription.fromString(orcSchema), orcConfig, batchSize);
+ }
+
+ /**
+ * Creates an OrcTableSouce from an ORC TypeDescription.
+ *
+ * @param path The path to read the ORC files from.
+ * @param orcSchema The schema of the ORC files as TypeDescription.
+ * @param orcConfig The configuration to read the ORC files.
+ * @param batchSize The number of Rows to read in a batch, default is 1000.
+ */
+ public OrcTableSource(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize) {
+ this(path, orcSchema, orcConfig, batchSize, null, null);
+ }
+
+ private OrcTableSource(String path, TypeDescription orcSchema, Configuration orcConfig,
+ int batchSize, int[] selectedFields, Predicate[] predicates) {
+
+ this.path = path;
+ this.orcSchema = orcSchema;
+ this.orcConfig = orcConfig;
+ this.batchSize = batchSize;
+ this.selectedFields = selectedFields;
+ this.predicates = predicates;
+
+ // determine the type information from the ORC schema
+ RowTypeInfo typeInfoFromSchema = (RowTypeInfo) OrcUtils.schemaToTypeInfo(this.orcSchema);
+
+ // set return type info
+ if (selectedFields == null) {
+ this.typeInfo = typeInfoFromSchema;
+ } else {
+ this.typeInfo = RowTypeInfo.projectFields(typeInfoFromSchema, selectedFields);
+ }
+
+ // create a TableSchema that corresponds to the ORC schema
+ this.tableSchema = new TableSchema(
+ typeInfoFromSchema.getFieldNames(),
+ typeInfoFromSchema.getFieldTypes()
+ );
+ }
+
+ @Override
+ public DataSet<Row> getDataSet(ExecutionEnvironment execEnv) {
+ OrcRowInputFormat orcIF = buildOrcInputFormat();
+ if (selectedFields != null) {
+ orcIF.selectFields(selectedFields);
+ }
+ if (predicates != null) {
+ for (OrcRowInputFormat.Predicate pred : predicates) {
+ orcIF.addPredicate(pred);
+ }
+ }
+ return execEnv.createInput(orcIF);
+ }
+
+ @VisibleForTesting
+ protected OrcRowInputFormat buildOrcInputFormat() {
+ return new OrcRowInputFormat(path, orcSchema, orcConfig, batchSize);
+ }
+
+ @Override
+ public TypeInformation<Row> getReturnType() {
+ return typeInfo;
+ }
+
+ @Override
+ public TableSchema getTableSchema() {
+ return this.tableSchema;
+ }
+
+ @Override
+ public TableSource<Row> projectFields(int[] selectedFields) {
+ // create a copy of the OrcTableSouce with new selected fields
+ return new OrcTableSource(path, orcSchema, orcConfig, batchSize, selectedFields, predicates);
+ }
+
+ @Override
+ public TableSource<Row> applyPredicate(List<Expression> predicates) {
+ ArrayList<Predicate> orcPredicates = new ArrayList<>();
+
+ // we do not remove any predicates from the list because ORC does not fully apply predicates
+ for (Expression pred : predicates) {
+ Predicate orcPred = toOrcPredicate(pred);
+ if (orcPred != null) {
+ orcPredicates.add(orcPred);
+ }
+ }
+
+ return new OrcTableSource(path, orcSchema, orcConfig, batchSize, selectedFields, orcPredicates.toArray(new Predicate[]{}));
+ }
+
+ @Override
+ public boolean isFilterPushedDown() {
+ return this.predicates != null;
+ }
+
+ @Override
+ public String explainSource() {
+ return "OrcFile[path=" + path + ", schema=" + orcSchema + ", filter=" + predicateString() + "]";
+ }
+
+ private String predicateString() {
+ if (predicates != null) {
+ return "AND(" + Arrays.toString(predicates) + ")";
+ } else {
+ return "TRUE";
+ }
+ }
+
+ // Predicate conversion for filter push-down.
+
+ private Predicate toOrcPredicate(Expression pred) {
+ if (pred instanceof Or) {
+ Predicate c1 = toOrcPredicate(((Or) pred).left());
+ Predicate c2 = toOrcPredicate(((Or) pred).right());
+ if (c1 == null || c2 == null) {
+ return null;
+ } else {
+ return new OrcRowInputFormat.Or(c1, c2);
+ }
+ } else if (pred instanceof Not) {
+ Predicate c = toOrcPredicate(((Not) pred).child());
+ if (c == null) {
+ return null;
+ } else {
+ return new OrcRowInputFormat.Not(c);
+ }
+ } else if (pred instanceof BinaryComparison) {
+
+ BinaryComparison binComp = (BinaryComparison) pred;
+
+ if (!isValid(binComp)) {
+ // not a valid predicate
+ return null;
+ }
+ PredicateLeaf.Type litType = getLiteralType(binComp);
+ if (litType == null) {
+ // unsupported literal type
+ return null;
+ }
+
+ boolean literalOnRight = literalOnRight(binComp);
+ String colName = getColumnName(binComp);
+ Serializable literal = (Serializable) getLiteral(binComp);
+
+ if (pred instanceof EqualTo) {
+ return new OrcRowInputFormat.Equals(colName, litType, literal);
+ } else if (pred instanceof NotEqualTo) {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.Equals(colName, litType, literal));
+ } else if (pred instanceof GreaterThan) {
+ if (literalOnRight) {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.LessThanEquals(colName, litType, literal));
+ } else {
+ return new OrcRowInputFormat.LessThan(colName, litType, literal);
+ }
+ } else if (pred instanceof GreaterThanOrEqual) {
+ if (literalOnRight) {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.LessThan(colName, litType, literal));
+ } else {
+ return new OrcRowInputFormat.LessThanEquals(colName, litType, literal);
+ }
+ } else if (pred instanceof LessThan) {
+ if (literalOnRight) {
+ return new OrcRowInputFormat.LessThan(colName, litType, literal);
+ } else {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.LessThanEquals(colName, litType, literal));
+ }
+ } else if (pred instanceof LessThanOrEqual) {
+ if (literalOnRight) {
+ return new OrcRowInputFormat.LessThanEquals(colName, litType, literal);
+ } else {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.LessThan(colName, litType, literal));
+ }
+ } else {
+ // unsupported predicate
+ return null;
+ }
+ } else if (pred instanceof UnaryExpression) {
+
+ UnaryExpression unary = (UnaryExpression) pred;
+ if (!isValid(unary)) {
+ // not a valid predicate
+ return null;
+ }
+ PredicateLeaf.Type colType = toOrcType(((UnaryExpression) pred).child().resultType());
+ if (colType == null) {
+ // unsupported type
+ return null;
+ }
+
+ String colName = getColumnName(unary);
+
+ if (pred instanceof IsNull) {
+ return new OrcRowInputFormat.IsNull(colName, colType);
+ } else if (pred instanceof IsNotNull) {
+ return new OrcRowInputFormat.Not(
+ new OrcRowInputFormat.IsNull(colName, colType));
+ } else if (pred instanceof IsTrue) {
+ if (colType == PredicateLeaf.Type.BOOLEAN) {
+ return new OrcRowInputFormat.Equals(colName, colType, Boolean.TRUE);
+ } else {
+ return null;
+ }
+ } else if (pred instanceof IsFalse) {
--- End diff --
Also add `IsNotTrue` and `IsNotFalse`?
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152532789
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalTableSourceScan.scala ---
@@ -62,7 +62,19 @@ class FlinkLogicalTableSourceScan(
override def computeSelfCost(planner: RelOptPlanner, metadata: RelMetadataQuery): RelOptCost = {
val rowCnt = metadata.getRowCount(this)
- planner.getCostFactory.makeCost(rowCnt, rowCnt, rowCnt * estimateRowSize(getRowType))
+
+ val adjustedCnt: Double = tableSource match {
+ case f: FilterableTableSource[_] if f.isFilterPushedDown =>
+ // ensure we prefer FilterableTableSources with pushed-down filters.
+ rowCnt - 1.0
--- End diff --
Shouldn't this be something like `rowCnt * 0.33`?
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152536536
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcUtils.java ---
@@ -0,0 +1,1511 @@
+/*
+ * 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.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+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.types.Row;
+
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+
+import org.apache.orc.TypeDescription;
+
+import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.function.DoubleFunction;
+import java.util.function.IntFunction;
+import java.util.function.LongFunction;
+
+/**
+ * A class that provides utility methods for orc file reading.
+ */
+class OrcUtils {
+
+ private static final long MILLIS_PER_DAY = 86400000; // = 24 * 60 * 60 * 1000
+ private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
+ /**
+ * Converts an ORC schema to a Flink TypeInformation.
+ *
+ * @param schema The ORC schema.
+ * @return The TypeInformation that corresponds to the ORC schema.
+ */
+ static TypeInformation schemaToTypeInfo(TypeDescription schema) {
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ return BasicTypeInfo.BOOLEAN_TYPE_INFO;
+ case BYTE:
+ return BasicTypeInfo.BYTE_TYPE_INFO;
+ case SHORT:
+ return BasicTypeInfo.SHORT_TYPE_INFO;
+ case INT:
+ return BasicTypeInfo.INT_TYPE_INFO;
+ case LONG:
+ return BasicTypeInfo.LONG_TYPE_INFO;
+ case FLOAT:
+ return BasicTypeInfo.FLOAT_TYPE_INFO;
+ case DOUBLE:
+ return BasicTypeInfo.DOUBLE_TYPE_INFO;
+ case DECIMAL:
+ return BasicTypeInfo.BIG_DEC_TYPE_INFO;
+ case STRING:
+ case CHAR:
+ case VARCHAR:
+ return BasicTypeInfo.STRING_TYPE_INFO;
+ case DATE:
+ return SqlTimeTypeInfo.DATE;
+ case TIMESTAMP:
+ return SqlTimeTypeInfo.TIMESTAMP;
+ case BINARY:
+ return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO;
+ case STRUCT:
+ List<TypeDescription> fieldSchemas = schema.getChildren();
+ TypeInformation[] fieldTypes = new TypeInformation[fieldSchemas.size()];
+ for (int i = 0; i < fieldSchemas.size(); i++) {
+ fieldTypes[i] = schemaToTypeInfo(fieldSchemas.get(i));
+ }
+ String[] fieldNames = schema.getFieldNames().toArray(new String[]{});
+ return new RowTypeInfo(fieldTypes, fieldNames);
+ case LIST:
+ TypeDescription elementSchema = schema.getChildren().get(0);
+ TypeInformation<?> elementType = schemaToTypeInfo(elementSchema);
+ // arrays of primitive types are handled as object arrays to support null values
+ return ObjectArrayTypeInfo.getInfoFor(elementType);
+ case MAP:
+ TypeDescription keySchema = schema.getChildren().get(0);
+ TypeDescription valSchema = schema.getChildren().get(1);
+ TypeInformation<?> keyType = schemaToTypeInfo(keySchema);
+ TypeInformation<?> valType = schemaToTypeInfo(valSchema);
+ return new MapTypeInfo<>(keyType, valType);
+ case UNION:
+ throw new UnsupportedOperationException("UNION type is not supported yet.");
+ default:
+ throw new IllegalArgumentException("Unknown type " + schema);
+ }
+ }
+
+ /**
+ * Fills an ORC batch into an array of Row.
+ *
+ * @param rows The batch of rows need to be filled.
+ * @param schema The schema of the ORC data.
+ * @param batch The ORC data.
+ * @param selectedFields The list of selected ORC fields.
+ * @return The number of rows that were filled.
+ */
+ static int fillRows(Row[] rows, TypeDescription schema, VectorizedRowBatch batch, int[] selectedFields) {
+
+ int rowsToRead = Math.min((int) batch.count(), rows.length);
+
+ List<TypeDescription> fieldTypes = schema.getChildren();
+ // read each selected field
+ for (int rowIdx = 0; rowIdx < selectedFields.length; rowIdx++) {
+ int orcIdx = selectedFields[rowIdx];
+ readField(rows, rowIdx, fieldTypes.get(orcIdx), batch.cols[orcIdx], null, rowsToRead);
+ }
+ return rowsToRead;
+ }
+
+ /**
+ * Reads a vector of data into an array of objects.
+ *
+ * @param vals The array that needs to be filled.
+ * @param fieldIdx If the vals array is an array of Row, the index of the field that needs to be filled.
+ * Otherwise a -1 must be passed and the data is directly filled into the array.
+ * @param schema The schema of the vector to read.
+ * @param vector The vector to read.
+ * @param lengthVector If the vector is of type List or Map, the number of sub-elements to read for each field. Otherwise, it must be null.
+ * @param childCount The number of vector entries to read.
+ */
+ private static void readField(Object[] vals, int fieldIdx, TypeDescription schema, ColumnVector vector, long[] lengthVector, int childCount) {
+
+ // check the type of the vector to decide how to read it.
+ switch (schema.getCategory()) {
+ case BOOLEAN:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readBoolean, OrcUtils::boolArray);
+ }
+ break;
+ case BYTE:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readByte, OrcUtils::byteArray);
+ }
+ break;
+ case SHORT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readShort, OrcUtils::shortArray);
+ }
+ break;
+ case INT:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readInt, OrcUtils::intArray);
+ }
+ break;
+ case LONG:
+ if (vector.noNulls) {
+ readNonNullLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ } else {
+ readLongColumn(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount, OrcUtils::readLong, OrcUtils::longArray);
+ }
+ break;
+ case FLOAT:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readFloat, OrcUtils::floatArray);
+ }
+ break;
+ case DOUBLE:
+ if (vector.noNulls) {
+ readNonNullDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ } else {
+ readDoubleColumn(vals, fieldIdx, (DoubleColumnVector) vector, lengthVector, childCount, OrcUtils::readDouble, OrcUtils::doubleArray);
+ }
+ break;
+ case CHAR:
+ case VARCHAR:
+ case STRING:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsString(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DATE:
+ if (vector.noNulls) {
+ readNonNullLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ } else {
+ readLongColumnAsDate(vals, fieldIdx, (LongColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case TIMESTAMP:
+ if (vector.noNulls) {
+ readNonNullTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ } else {
+ readTimestampColumn(vals, fieldIdx, (TimestampColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case BINARY:
+ if (vector.noNulls) {
+ readNonNullBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ } else {
+ readBytesColumnAsBinary(vals, fieldIdx, (BytesColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case DECIMAL:
+ if (vector.noNulls) {
+ readNonNullDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ else {
+ readDecimalColumn(vals, fieldIdx, (DecimalColumnVector) vector, lengthVector, childCount);
+ }
+ break;
+ case STRUCT:
+ if (vector.noNulls) {
+ readNonNullStructColumn(vals, fieldIdx, (StructColumnVector) vector, schema, lengthVector, childCount);
+ } else {
+ readStructColumn(vals, fieldIdx, (StructColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case LIST:
+ if (vector.noNulls) {
+ readNonNullListColumn(vals, fieldIdx, (ListColumnVector) vector, schema, lengthVector, childCount);
+ }
+ else {
+ readListColumn(vals, fieldIdx, (ListColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case MAP:
+ if (vector.noNulls) {
+ readNonNullMapColumn(vals, fieldIdx, (MapColumnVector) vector, schema, lengthVector, childCount);
+ }
+ else {
+ readMapColumn(vals, fieldIdx, (MapColumnVector) vector, schema, lengthVector, childCount);
+ }
+ break;
+ case UNION:
--- End diff --
No, not really. I implemented what was available from the API, i.e., `TypeDescriptor`.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152519868
--- Diff: flink-core/src/main/java/org/apache/flink/api/java/typeutils/RowTypeInfo.java ---
@@ -355,4 +355,21 @@ public void addComparatorField(int fieldId, TypeComparator<?> comparator) {
comparatorOrders);
}
}
+
+ /**
+ * Creates a {@link RowTypeInfo} with projected fields.
+ *
+ * @param rowType The original RowTypeInfo whose fields are projected
+ * @param fieldMapping The field mapping of the projection
+ * @return A RowTypeInfo with projected fields.
+ */
+ public static RowTypeInfo projectFields(RowTypeInfo rowType, int[] fieldMapping) {
--- End diff --
Why is this method static?
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152535417
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcRowInputFormat.java ---
@@ -0,0 +1,747 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.StripeInformation;
+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 java.io.Serializable;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.orc.OrcUtils.fillRows;
+
+/**
+ * InputFormat to read ORC files.
+ */
+public class OrcRowInputFormat extends FileInputFormat<Row> implements ResultTypeQueryable<Row> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrcRowInputFormat.class);
+ // the number of rows read in a batch
+ private static final int DEFAULT_BATCH_SIZE = 1000;
+
+ // the number of fields rows to read in a batch
+ private int batchSize;
+ // the configuration to read with
+ private Configuration conf;
--- End diff --
Yes, that's why we override `writeObject` and `readObject`.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152579106
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcRowInputFormat.java ---
@@ -0,0 +1,747 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.StripeInformation;
+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 java.io.Serializable;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.orc.OrcUtils.fillRows;
+
+/**
+ * InputFormat to read ORC files.
+ */
+public class OrcRowInputFormat extends FileInputFormat<Row> implements ResultTypeQueryable<Row> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrcRowInputFormat.class);
+ // the number of rows read in a batch
+ private static final int DEFAULT_BATCH_SIZE = 1000;
+
+ // the number of fields rows to read in a batch
+ private int batchSize;
+ // the configuration to read with
+ private Configuration conf;
+ // the schema of the ORC files to read
+ private TypeDescription schema;
+
+ // the fields of the ORC schema that the returned Rows are composed of.
+ private int[] selectedFields;
+ // the type information of the Rows returned by this InputFormat.
+ private transient RowTypeInfo rowType;
+
+ // the ORC reader
+ private transient RecordReader orcRowsReader;
+ // the vectorized row data to be read in a batch
+ private transient VectorizedRowBatch rowBatch;
+ // the vector of rows that is read in a batch
+ private transient Row[] rows;
+
+ // the number of rows in the current batch
+ private transient int rowsInBatch;
+ // the index of the next row to return
+ private transient int nextRow;
+
+ private ArrayList<Predicate> conjunctPredicates = new ArrayList<>();
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig, int batchSize) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, batchSize);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param orcSchema The schema of the ORC files as ORC TypeDescription.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize) {
+ super(new Path(path));
+
+ // configure OrcInputFormat
+ this.schema = orcSchema;
+ this.rowType = (RowTypeInfo) OrcUtils.schemaToTypeInfo(schema);
+ this.conf = orcConfig;
+ this.batchSize = batchSize;
+
+ // set default selection mask, i.e., all fields.
+ this.selectedFields = new int[this.schema.getChildren().size()];
+ for (int i = 0; i < selectedFields.length; i++) {
+ this.selectedFields[i] = i;
+ }
+ }
+
+ /**
+ * Adds a filter predicate to reduce the number of rows to be returned by the input format.
+ * Multiple conjunctive predicates can be added by calling this method multiple times.
+ *
+ * <p>Note: Predicates can significantly reduce the amount of data that is read.
+ * However, the OrcRowInputFormat does not guarantee that all returned rows qualitfy the
+ * predicates. Moreover, predicates are only applied if the referenced field is among the
+ * selected fields.</p>
+ *
+ * @param predicate The filter predicate.
+ */
+ public void addPredicate(Predicate predicate) {
+ // validate
+ validatePredicate(predicate);
+ // add predicate
+ this.conjunctPredicates.add(predicate);
+ }
+
+ private void validatePredicate(Predicate pred) {
+ if (pred instanceof ColumnPredicate) {
+ // check column name
+ String colName = ((ColumnPredicate) pred).columnName;
+ if (!this.schema.getFieldNames().contains(colName)) {
+ throw new IllegalArgumentException("Predicate cannot be applied. " +
+ "Column '" + colName + "' does not exist in ORC schema.");
+ }
+ } else if (pred instanceof Not) {
+ validatePredicate(((Not) pred).child());
+ } else if (pred instanceof Or) {
+ for (Predicate p : ((Or) pred).children()) {
+ validatePredicate(p);
+ }
+ }
+ }
+
+ /**
+ * Selects the fields from the ORC schema that are returned by InputFormat.
+ *
+ * @param selectedFields The indices of the fields of the ORC schema that are returned by the InputFormat.
+ */
+ public void selectFields(int... selectedFields) {
+ // set field mapping
+ this.selectedFields = selectedFields;
+ // adapt result type
+ this.rowType = RowTypeInfo.projectFields(this.rowType, selectedFields);
+ }
+
+ /**
+ * Computes the ORC projection mask of the fields to include from the selected fields.rowOrcInputFormat.nextRecord(null).
+ *
+ * @return The ORC projection mask.
+ */
+ private boolean[] computeProjectionMask() {
+ // mask with all fields of the schema
+ boolean[] projectionMask = new boolean[schema.getMaximumId() + 1];
+ // for each selected field
+ for (int inIdx : selectedFields) {
+ // set all nested fields of a selected field to true
+ 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();
+ // create and initialize the row batch
+ this.rows = new Row[batchSize];
+ for (int i = 0; i < batchSize; i++) {
+ rows[i] = new Row(selectedFields.length);
+ }
+ }
+
+ @Override
+ public void open(FileInputSplit fileSplit) throws IOException {
+
+ LOG.debug("Opening ORC file {}", fileSplit.getPath());
+
+ // open ORC file and create reader
+ org.apache.hadoop.fs.Path hPath = new org.apache.hadoop.fs.Path(fileSplit.getPath().getPath());
+ Reader orcReader = OrcFile.createReader(hPath, OrcFile.readerOptions(conf));
+
+ // get offset and length for the stripes that start in the split
+ Tuple2<Long, Long> offsetAndLength = getOffsetAndLengthForSplit(fileSplit, getStripes(orcReader));
+
+ // create ORC row reader configuration
+ Reader.Options options = getOptions(orcReader)
+ .schema(schema)
+ .range(offsetAndLength.f0, offsetAndLength.f1)
+ .useZeroCopy(OrcConf.USE_ZEROCOPY.getBoolean(conf))
+ .skipCorruptRecords(OrcConf.SKIP_CORRUPT_DATA.getBoolean(conf))
+ .tolerateMissingSchema(OrcConf.TOLERATE_MISSING_SCHEMA.getBoolean(conf));
+
+ // configure filters
+ if (!conjunctPredicates.isEmpty()) {
+ SearchArgument.Builder b = SearchArgumentFactory.newBuilder();
+ b = b.startAnd();
+ for (Predicate predicate : conjunctPredicates) {
+ predicate.add(b);
+ }
+ b = b.end();
+ options.searchArgument(b.build(), new String[]{});
+ }
+
+ // configure selected fields
+ options.include(computeProjectionMask());
+
+ // create ORC row reader
+ this.orcRowsReader = orcReader.rows(options);
+
+ // assign ids
+ this.schema.getId();
+ // create row batch
+ this.rowBatch = schema.createRowBatch(batchSize);
+ rowsInBatch = 0;
+ nextRow = 0;
+ }
+
+ @VisibleForTesting
+ Reader.Options getOptions(Reader orcReader) {
+ return orcReader.options();
+ }
+
+ @VisibleForTesting
+ List<StripeInformation> getStripes(Reader orcReader) {
+ return orcReader.getStripes();
+ }
+
+ private Tuple2<Long, Long> getOffsetAndLengthForSplit(FileInputSplit split, List<StripeInformation> stripes) {
+ long splitStart = split.getStart();
+ long splitEnd = splitStart + split.getLength();
+
+ long readStart = Long.MAX_VALUE;
+ long readEnd = Long.MIN_VALUE;
+
+ for (StripeInformation s : stripes) {
+ if (splitStart <= s.getOffset() && s.getOffset() < splitEnd) {
+ // stripe starts in split, so it is included
+ readStart = Math.min(readStart, s.getOffset());
+ readEnd = Math.max(readEnd, s.getOffset() + s.getLength());
+ }
+ }
+
+ if (readStart < Long.MAX_VALUE) {
+ // at least one split is included
+ return Tuple2.of(readStart, readEnd - readStart);
+ } else {
+ return Tuple2.of(0L, 0L);
+ }
+ }
+
+ @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();
+ }
+
+ /**
+ * Checks if there is at least one row left in the batch to return.
+ * If no more row are available, it reads another batch of rows.
+ *
+ * @return Returns true if there is one more row to return, false otherwise.
+ * @throws IOException throw if an exception happens while reading a batch.
+ */
+ private boolean ensureBatch() throws IOException {
+
+ if (nextRow >= rowsInBatch) {
+ // No more rows available in the Rows array.
+ nextRow = 0;
+ // Try to read the next batch if rows from the ORC file.
+ boolean moreRows = orcRowsReader.nextBatch(rowBatch);
+
+ if (moreRows) {
+ // Load the data into the Rows array.
+ rowsInBatch = fillRows(rows, schema, rowBatch, selectedFields);
+ }
+ return moreRows;
+ }
+ // there is at least one Row left in the Rows array.
+ return true;
+ }
+
+ @Override
+ public Row nextRecord(Row reuse) throws IOException {
+ // return the next row
+ return rows[this.nextRow++];
+ }
+
+ @Override
+ public TypeInformation<Row> getProducedType() {
+ return rowType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Custom serialization methods
+ // --------------------------------------------------------------------------------------------
+
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.writeInt(batchSize);
+ this.conf.write(out);
+ out.writeUTF(schema.toString());
+
+ out.writeInt(selectedFields.length);
+ for (int f : selectedFields) {
+ out.writeInt(f);
+ }
+
+ out.writeInt(conjunctPredicates.size());
+ for (Predicate p : conjunctPredicates) {
+ out.writeObject(p);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ batchSize = in.readInt();
+ org.apache.hadoop.conf.Configuration configuration = new org.apache.hadoop.conf.Configuration();
+ configuration.readFields(in);
+
+ if (this.conf == null) {
+ this.conf = configuration;
+ }
+ this.schema = TypeDescription.fromString(in.readUTF());
+
+ this.selectedFields = new int[in.readInt()];
+ for (int i = 0; i < selectedFields.length; i++) {
+ this.selectedFields[i] = in.readInt();
+ }
+
+ this.conjunctPredicates = new ArrayList<>();
+ int numPreds = in.readInt();
+ for (int i = 0; i < numPreds; i++) {
+ conjunctPredicates.add((Predicate) in.readObject());
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Classes to define predicates
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A filter predicate that can be evaluated by the OrcRowInputFormat.
+ */
+ public abstract static class Predicate implements Serializable {
+ protected abstract SearchArgument.Builder add(SearchArgument.Builder builder);
+ }
+
+ abstract static class ColumnPredicate extends Predicate {
+ final String columnName;
+ final PredicateLeaf.Type literalType;
+
+ ColumnPredicate(String columnName, PredicateLeaf.Type literalType) {
+ this.columnName = columnName;
+ this.literalType = literalType;
+ }
+
+ Object castLiteral(Serializable literal) {
+
+ switch (literalType) {
+ case LONG:
+ if (literal instanceof Byte) {
+ return new Long((Byte) literal);
+ } else if (literal instanceof Short) {
+ return new Long((Short) literal);
+ } else if (literal instanceof Integer) {
+ return new Long((Integer) literal);
+ } else if (literal instanceof Long) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a LONG column requires an integer " +
+ "literal, i.e., Byte, Short, Integer, or Long.");
+ }
+ case FLOAT:
+ if (literal instanceof Float) {
+ return new Double((Float) literal);
+ } else if (literal instanceof Double) {
+ return literal;
+ } else if (literal instanceof BigDecimal) {
+ return ((BigDecimal) literal).doubleValue();
+ } else {
+ throw new IllegalArgumentException("A predicate on a FLOAT column requires a floating " +
+ "literal, i.e., Float or Double.");
+ }
+ case STRING:
+ if (literal instanceof String) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a STRING column requires a floating " +
+ "literal, i.e., Float or Double.");
+ }
+ case BOOLEAN:
+ if (literal instanceof Boolean) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a BOOLEAN column requires a Boolean literal.");
+ }
+ case DATE:
+ if (literal instanceof Date) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a DATE column requires a java.sql.Date literal.");
+ }
+ case TIMESTAMP:
+ if (literal instanceof Timestamp) {
+ return literal;
+ } else {
+ throw new IllegalArgumentException("A predicate on a TIMESTAMP column requires a java.sql.Timestamp literal.");
+ }
+ case DECIMAL:
+ if (literal instanceof BigDecimal) {
+ return new HiveDecimalWritable(HiveDecimal.create((BigDecimal) literal));
+ } else {
+ throw new IllegalArgumentException("A predicate on a DECIMAL column requires a BigDecimal literal.");
+ }
+ default:
+ throw new IllegalArgumentException("Unknown literal type " + literalType);
+ }
+ }
+ }
+
+ abstract static class BinaryPredicate extends ColumnPredicate {
+ final Serializable literal;
+
+ BinaryPredicate(String columnName, PredicateLeaf.Type literalType, Serializable literal) {
+ super(columnName, literalType);
+ this.literal = literal;
+ }
+ }
+
+ /**
+ * An EQUALS predicate that can be evaluated by the OrcRowInputFormat.
+ */
+ public static class Equals extends BinaryPredicate {
+ /**
+ * Creates an EQUALS predicate.
+ *
+ * @param columnName The column to check.
+ * @param literalType The type of the literal.
+ * @param literal The literal value to check the column against.
+ */
+ public Equals(String columnName, PredicateLeaf.Type literalType, Serializable literal) {
+ super(columnName, literalType, literal);
+ }
+
+ @Override
+ protected SearchArgument.Builder add(SearchArgument.Builder builder) {
+ return builder.equals(columnName, literalType, castLiteral(literal));
+ }
+
+ @Override
+ public String toString() {
+ return columnName + " = " + literal;
+ }
+ }
+
+ /**
+ * An EQUALS predicate that can be evaluated with Null safety by the OrcRowInputFormat.
+ */
+ public static class NullSafeEquals extends BinaryPredicate {
+ /**
+ * Creates a null-safe EQUALS predicate.
+ *
+ * @param columnName The column to check.
+ * @param literalType The type of the literal.
+ * @param literal The literal value to check the column against.
+ */
+ public NullSafeEquals(String columnName, PredicateLeaf.Type literalType, Serializable literal) {
--- End diff --
An untested interface class ;-)
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152634076
--- Diff: flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceITCase.java ---
@@ -0,0 +1,112 @@
+/*
+ * 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.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.BatchTableEnvironment;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for {@link OrcTableSource}.
+ */
+public class OrcTableSourceITCase extends MultipleProgramsTestBase {
--- End diff --
`TableProgramsTestBase` doesn't really give any advantage over `MultipleProgramsTestBase` since I'm testing the table source and not the null-behavior of the query execution.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152502704
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcRowInputFormat.java ---
@@ -0,0 +1,747 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.StripeInformation;
+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 java.io.Serializable;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.orc.OrcUtils.fillRows;
+
+/**
+ * InputFormat to read ORC files.
+ */
+public class OrcRowInputFormat extends FileInputFormat<Row> implements ResultTypeQueryable<Row> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrcRowInputFormat.class);
+ // the number of rows read in a batch
+ private static final int DEFAULT_BATCH_SIZE = 1000;
+
+ // the number of fields rows to read in a batch
+ private int batchSize;
+ // the configuration to read with
+ private Configuration conf;
--- End diff --
`Configuration` is not serializable.
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152499605
--- Diff: docs/dev/table/sourceSinks.md ---
@@ -485,6 +486,50 @@ val csvTableSource = CsvTableSource
{% top %}
+### OrcTableSource
+
+The `OrcTableSource` reads [ORC files](https://orc.apache.org). ORC is a file format for structured data and stores the data in a compressed, columnar representation. ORC is very storage efficient and supports projection and filter push-down.
+
+An `OrcTableSource` is created as shown below:
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+
+// create Hadoop Configuration
+Configuration config = new Configuration();
+
+OrcTableSource orcTableSource = new OrcTableSource(
--- End diff --
Should we also provide a builder as it is done for all other sources to be consistent?
---
[GitHub] flink pull request #5043: [FLINK-2170] [connectors] Add OrcRowInputFormat an...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5043#discussion_r152518885
--- Diff: flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcRowInputFormat.java ---
@@ -0,0 +1,747 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.StripeInformation;
+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 java.io.Serializable;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.orc.OrcUtils.fillRows;
+
+/**
+ * InputFormat to read ORC files.
+ */
+public class OrcRowInputFormat extends FileInputFormat<Row> implements ResultTypeQueryable<Row> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrcRowInputFormat.class);
+ // the number of rows read in a batch
+ private static final int DEFAULT_BATCH_SIZE = 1000;
+
+ // the number of fields rows to read in a batch
+ private int batchSize;
+ // the configuration to read with
+ private Configuration conf;
+ // the schema of the ORC files to read
+ private TypeDescription schema;
+
+ // the fields of the ORC schema that the returned Rows are composed of.
+ private int[] selectedFields;
+ // the type information of the Rows returned by this InputFormat.
+ private transient RowTypeInfo rowType;
+
+ // the ORC reader
+ private transient RecordReader orcRowsReader;
+ // the vectorized row data to be read in a batch
+ private transient VectorizedRowBatch rowBatch;
+ // the vector of rows that is read in a batch
+ private transient Row[] rows;
+
+ // the number of rows in the current batch
+ private transient int rowsInBatch;
+ // the index of the next row to return
+ private transient int nextRow;
+
+ private ArrayList<Predicate> conjunctPredicates = new ArrayList<>();
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, DEFAULT_BATCH_SIZE);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param schemaString The schema of the ORC files as String.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, String schemaString, Configuration orcConfig, int batchSize) {
+ this(path, TypeDescription.fromString(schemaString), orcConfig, batchSize);
+ }
+
+ /**
+ * Creates an OrcRowInputFormat.
+ *
+ * @param path The path to read ORC files from.
+ * @param orcSchema The schema of the ORC files as ORC TypeDescription.
+ * @param orcConfig The configuration to read the ORC files with.
+ * @param batchSize The number of Row objects to read in a batch.
+ */
+ public OrcRowInputFormat(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize) {
+ super(new Path(path));
+
+ // configure OrcInputFormat
+ this.schema = orcSchema;
+ this.rowType = (RowTypeInfo) OrcUtils.schemaToTypeInfo(schema);
+ this.conf = orcConfig;
+ this.batchSize = batchSize;
+
+ // set default selection mask, i.e., all fields.
+ this.selectedFields = new int[this.schema.getChildren().size()];
+ for (int i = 0; i < selectedFields.length; i++) {
+ this.selectedFields[i] = i;
+ }
+ }
+
+ /**
+ * Adds a filter predicate to reduce the number of rows to be returned by the input format.
+ * Multiple conjunctive predicates can be added by calling this method multiple times.
+ *
+ * <p>Note: Predicates can significantly reduce the amount of data that is read.
+ * However, the OrcRowInputFormat does not guarantee that all returned rows qualitfy the
+ * predicates. Moreover, predicates are only applied if the referenced field is among the
+ * selected fields.</p>
--- End diff --
Remove `</p>`
---