This PR adds Hive (https://github.com/apache/hive) and Presto (https://github.com/prestodb/presto) support for parquet messages written with ProtoParquetWriter. Hive and other tools, such as Presto (used by AWS Athena), rely on specific LIST/MAP wrappers (as defined in the parquet spec: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md). These wrappers are currently missing from the ProtoParquet schema. AvroParquet works just fine, because it adds these wrappers when it deals with arrays and maps. This PR brings these wrappers in parquet-proto, providing the same functionality that already exists in parquet-avro.

This is backward compatible. Messages written without the extra LIST/MAP wrappers are still being read successfully using the updated ProtoParquetReader.

Regarding the change. Given the following protobuf schema:

message ListOfPrimitives {
    repeated int64 my_repeated_id = 1;
}

Old parquet schema was:

message ListOfPrimitives {
  repeated int64 my_repeated_id = 1;
}

New parquet schema is:

message ListOfPrimitives {
  required group my_repeated_id (LIST) = 1 {
    repeated group list {
      required int64 element;
    }
  }
}

For list of messages, the changes look like this:

Protobuf schema:

message ListOfMessages {
    string top_field = 1;
    repeated MyInnerMessage first_array = 2;
}

message MyInnerMessage {
    int32 inner_field = 1;
}

Old parquet schema was:

message TestProto3.ListOfMessages {
  optional binary top_field (UTF8) = 1;
  repeated group first_array = 2 {
    optional int32 inner_field = 1;
  }
}

The expected parquet schema, compatible with Hive (and similar to parquet-avro) is the following (notice the LIST wrapper):

message TestProto3.ListOfMessages {
  optional binary top_field (UTF8) = 1;
  required group first_array (LIST) = 2 {
    repeated group list {
      optional group element {
        optional int32 inner_field = 1;
      }
    }
  }
}

Similar for maps. Protobuf schema:

message TopMessage {
    map<int64, MyInnerMessage> myMap = 1;
}

message MyInnerMessage {
    int32 inner_field = 1;
}

Old parquet schema:

message TestProto3.TopMessage {
  repeated group myMap = 1 {
    optional int64 key = 1;
    optional group value = 2 {
      optional int32 inner_field = 1;
    }
  }
}

New parquet schema (notice the MAP wrapper):

message TestProto3.TopMessage {
  required group myMap (MAP) = 1 {
    repeated group key_value {
      required int64 key;
      optional group value {
        optional int32 inner_field = 1;
      }
    }
  }
}

Jira: https://issues.apache.org/jira/browse/PARQUET-968

Currently, ordering of Binary in Parquet is based on byte-by-byte comparison. This doesn't match the standard method of lexicographic sorting of Unicode strings, you can see an example of this in

• Avro: https://github.com/apache/avro/blob/master/lang/java/avro/src/main/java/org/apache/avro/io/BinaryData.java#L184
• Spark: https://github.com/apache/spark/blob/master/common/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java#L835

This overrides comparison on FromStringBinary to implement correct sort order.

@julienledem

Context:

Parquet is currently structured to choose the appropriate value writer based on the type of the column as well as the Parquet version. As of now, the writer(s) (and hence encoding) for each data type is hard coded in the Parquet source code.

This PR adds support for being able to override the encodings per type via config. That allows users to experiment with various encoding strategies manually as well as enables them to override the hardcoded defaults if they don't suit their use case.

We can override encodings per data type (int32 / int64 / ...). Something on the lines of:

parquet.writer.encoding-override.<type> = "encoding1[,encoding2]"

As an example:

"parquet.writer.encoding-override.int32" = "plain"
(Chooses Plain encoding and hence the PlainValuesWriter).

When a primary + fallback need to be specified, we can do the following:

"parquet.writer.encoding-override.binary" = "rle_dictionary,delta_byte_array"
(Chooses RLE_DICTIONARY encoding as the initial encoding and DELTA_BYTE_ARRAY encoding as the fallback and hence creates a FallbackWriter(PlainBinaryDictionaryValuesWriter, DeltaByteArrayWriter). 

In such cases we can mandate that the first encoding listed must allow for Fallbacks by implementing RequiresFallback.

PR notes:

• Restructured the ValuesWriter creation code. Pulled it out of ParquetProperties into a new class and refactored the flow based on type as it was getting hard to follow and I felt adding the overrides would make it harder. Added a bunch of unit tests to verify the ValuesWriter we create for combinations of type, parquet version and dictionary on / off.
• Added unit tests to verify parsing of the encoding overrides + creation of ValuesWriters based on these overrides.
• Manually tested some encoding overrides scenarios out on Hadoop (both parquet v1, v2).

Currently for creating a user defined predicate using the new filter api, no value can be passed to create a dynamic filter at runtime. This reduces the usefulness of the user defined predicate, and meaningful predicates cannot be created. We can add a generic Object value that is passed through the api, which can internally be used in the keep function of the user defined predicate for creating many different types of filters. For example, in spark sql, we can pass in a list of filter values for a where IN clause query and filter the row values based on that list.

This PR addresses this issue: https://issues.apache.org/jira/browse/PARQUET-1633

I have not added unit tests, as to check overflow conditions I would need test data over 2GB in size (on disk, compressed), considerably larger in-memory and thus requiring significant CI resources.

The issue was using an int for length field, which for parquet files with very large row_group_size (row groups over 2GB) would cause silent integer overflow, manifesting itself as negative length and an attempt to create an ArrayList with negative length.

I have tested reading a 6.6GB parquet file with a huge row group size (works out at over 2GB) which recreated this problem, and with this modification can read the file without any issues.

Parquet Avro reader can't convert such Parquet schemas, where a group field name is reused in an inner structure. The converter creates Avro record schma in this case, but in Avro record types should have a unique name, therefore the result is an invalid Avro schema. This patch fixes this case by adding a namespace for the record if the name was defined before, this way making the record names unique.

This fixes PARQUET-400 by replacing CompatibilityUtil with SeekableInputStream that's implemented for hadoop-1 and hadoop-2. The benefit of this approach is that SeekableInputStream can be used for non-Hadoop file systems in the future.

This also changes the default Hadoop version to Hadoop-2. The library is still compatible with Hadoop 1.x, but this makes building Hadoop-2 classes, like H2SeekableInputStream, much easier and removes the need for multiple hadoop versions during compilation.

The current Log class is intended to allow swapping out logger back-ends, but SLF4J already does this. It also doesn't expose as nice of an API as SLF4J, which can handle formatting to avoid the cost of building log messages that won't be used.

We should deprecate the org.apache.parquet.Log class and move to using SLF4J directly, instead of wrapping SLF4J (PARQUET-305).

…ncryption

Make sure you have checked all steps below.

Jira

• [ ] My PR addresses the following Parquet Jira issues and references them in the PR title. For example, "PARQUET-1234: My Parquet PR"
  • https://issues.apache.org/jira/browse/PARQUET-XXX
  • In case you are adding a dependency, check if the license complies with the ASF 3rd Party License Policy.

Tests

• [ ] My PR adds the following unit tests OR does not need testing for this extremely good reason:

Commits

• [ ] My commits all reference Jira issues in their subject lines. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters (not including Jira issue reference)
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• [ ] In case of new functionality, my PR adds documentation that describes how to use it.
  • All the public functions and the classes in the PR contain Javadoc that explain what it does

JIRA: https://issues.apache.org/jira/browse/PARQUET-225

As of now, parquet doesn't support delta encoding for INT64. However it is planned in the format: https://github.com/Parquet/parquet-format/blob/master/Encodings.md

The benefits of this feature are huge. For timestamps it achieves twice better compression than SNAPPY on plain encoding, and the reading is faster. This feature is actually advertised on the home page of Parquet, even though it is not yet implemented: http://parquet.incubator.apache.org/ http://image.slidesharecdn.com/hadoopsummit-140630160016-phpapp01/95/efficient-data-storage-for-analytics-with-apache-parquet-20-30-638.jpg?cb=1404162126

This pull request adds full support for INT64. It keeps INT32 binary unchanged, even though it reads it with the INT64 implementation. That might be slower on 32 bit processors, but I don't not consider it a problem.

Regarding testing, all test made for INT32 are ran against the new INT64 implementation and also 2 new unit tests are added specifically for INT64.

The code in parquet-cascading is adapted to the API as of Cascading 2.5.3

Some incompatible changes were introduced in Cascading 3.0. This patch forks the parquet-cascading module to also provide a parquet-cascading3 module, which is about identical save for overloads which changed from requiring a Foo to requiring a Foo<? extends JobConf>

Make sure you have checked all steps below.

Jira

• [ ] My PR addresses the following Parquet Jira issues and references them in the PR title. For example, "PARQUET-1234: My Parquet PR"
  • https://issues.apache.org/jira/browse/PARQUET-XXX
  • In case you are adding a dependency, check if the license complies with the ASF 3rd Party License Policy.

Tests

• [ ] My PR adds the following unit tests OR does not need testing for this extremely good reason:

Commits

• [ ] My commits all reference Jira issues in their subject lines. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters (not including Jira issue reference)
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• [ ] In case of new functionality, my PR adds documentation that describes how to use it.
  • All the public functions and the classes in the PR contain Javadoc that explain what it does

Jira

• This patch aims to solve the first step of PARQUET-2075.

Tests

• Make sure all tasks pass, especially CompressionConverterTest, ColumnPrunerTest, ColumnMaskerTest, and ColumnEncryptorTest.

Commits

• A new ParquetRewriter is introduced to unify rewriting logic.
• RewriteOptions is defined to provide essential settings.
• CompressionConverter, ColumnPruner, ColumnMasker, and ColumnEncryptor have been refactored and marked as deprecated.

now that we have 0.17 to use

Make sure you have checked all steps below.

Jira

• [ ] My PR addresses the following Parquet Jira issues and references them in the PR title. For example, "PARQUET-1234: My Parquet PR"
  • https://issues.apache.org/jira/browse/PARQUET-XXX
  • In case you are adding a dependency, check if the license complies with the ASF 3rd Party License Policy.

Tests

• [x] My PR adds the following unit tests OR does not need testing for this extremely good reason:

Commits

• [ ] My commits all reference Jira issues in their subject lines. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters (not including Jira issue reference)
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• [ ] In case of new functionality, my PR adds documentation that describes how to use it.
  • All the public functions and the classes in the PR contain Javadoc that explain what it does

The PR includes 3 aspects：

1. Use java17 vector api to decode bit-packing , the performance gain is 4x ~ 8x according to the microbenchmark
2. Upgrade the project to java17 to support java vector api
3. Add ParquetReadRouter to compatible with different platform when computing engines(such as spark) read parquet.

Jira

• https://issues.apache.org/jira/projects/PARQUET/issues/PARQUET-2159?filter=allopenissues

Tests

• Add unit tests org.apache.parquet.column.values.bitpacking.TestByteBitPackingVectorLE

Commits

• [ ] My commits all reference Jira issues in their subject lines. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters (not including Jira issue reference)
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• The PR adds maven profile vector to enable generate java17 vector bit-packing decode opt. code, and run junit tests: mvn clean install -P vector
• The PR needs Intel Ice Lake CPU to run junit tests
• The contributors are jiangjiguang jiyu1021 guangzegu Fang-Xie , and co-contributor is jatin-bhateja

This adds a new API, InputFile.newStream, that allow specifying a range for the returned input stream. This allows the implementations (especially for cloud storages like S3) to only download the data within the specified range, instead of the whole object.

Make sure you have checked all steps below.

Jira

• [ ] My PR addresses the following Parquet Jira issues and references them in the PR title. For example, "PARQUET-1234: My Parquet PR"
  • https://issues.apache.org/jira/browse/PARQUET-XXX
  • In case you are adding a dependency, check if the license complies with the ASF 3rd Party License Policy.

Tests

• [ ] My PR adds the following unit tests OR does not need testing for this extremely good reason:

Commits

• [ ] My commits all reference Jira issues in their subject lines. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters (not including Jira issue reference)
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• [ ] In case of new functionality, my PR adds documentation that describes how to use it.
  • All the public functions and the classes in the PR contain Javadoc that explain what it does

The PR adds the new ByteBuffer api and also updates ColumnChunkPageReadStore.readPage to use the new API. A few additional classes were touched (ParquetReader.Builder, BytesInput) to allow an allocator to be specified and/or to avoid ByteBuffer -> byte array copying. These changes were necessary to enable the unit test. A user option has been added to explicitly enable/disable the use of the ByteBuffer api for decryption.

Jira

• My PR addresses t Parquet 2212

Tests

• Updates Unit test(s) in org.apache.parquet.crypto.TestPropertiesDrivenEncryption