apache / apache/parquet-java

mandatory status of avro columns ignored

Đang mở
#1,959 0 bình luận 0 reaction 0 người được giao Xem trên GitHub
Component: Avro Component: Parquet Priority: Major Type: bug
Ngôn ngữ chính
Java
Star
3.1k
Fork
1.6k
Merge trung bình
3 ngày 12 giờ
Pull request đã merge (30 ngày)
33

Mô tả

Avro spec schema [resolution rules ](https://avro.apache.org/docs/1.7.7/spec.html#schema_record) say:
"if the reader's record schema has a field with no default value, and writer's schema does not have a field with the same name, an error is signalled."

I can't find the implementation of this aspect in parquet.avro and indeed observe this rule seemingly ignored. I am using 1.6.0 because that's what we can get off maven.

My writer's schema:
```Java
{
"type" : "record",
"name" : "SampleSchema_v1",
"namespace" : "com.xxxx.spark",
"fields" : [ {
"name" : "stringField",
"type" : "string",
"doc" : "Sample string field"
},{
"name" : "longField",
"type" : "long",
"doc" : "Sample long field"
} ],
"doc:" : "A sample/test schema"
}
```

My reader schema:
```Java
{
"type" : "record",
"name" : "SampleSchema_newDefaultlessCol",
"namespace" : "com.xxxx.spark",
"fields" : [ {
"name" : "stringField",
"type" : "string",
"doc" : "Sample string field"
},{
"name" : "longField",
"type" : "long",
"doc" : "Sample long field"
},{
"name" : "mandatoryIntField",
"type" : "int",
"doc" : "Sample mandatory! int field"
}],
"doc:" : "v1 + one extra column that has no default"
}
```
This is my test case:
```Java
"accept new column w/o a default [schema-evolution, undesired]" in new MockAvroParquetGrid {
//TODO: the behaviour this test case exercises is UNDESIRED, i.e.: a new column with no default value should
//TODO: Ticket to track this: https://jira.xxxx.io/browse/ADR-610
//constitute an incompatible schema break, instead, this thing uses 0 for the default
val inputSampleRecordsV1 = Seq(new SampleSchema_v1(s"string", 1))
dao.writeParquet[SampleSchema_v1](
SparkBase.sc.parallelize(inputSampleRecordsV1),
SampleSchema_v1.SCHEMA$,
parquetFolder
)

dao
.readParquet[SampleSchema_newDefaultlessCol](parquetFolder, SampleSchema_newDefaultlessCol.SCHEMA$)
.collect().toSeq.head
.getMandatoryIntField must equalTo(0) //TODO: zero is an unwelcome guess
}
```
This is the implementation of writeParquet and readParquet
```Java
def writeParquet[C](source: RDD[C], schema: org.apache.avro.Schema, dstPath: String)
(implicit ctag: ClassTag[C]): Unit = {
val hadoopJob = Job.getInstance()
ParquetOutputFormat.setWriteSupportClass(hadoopJob, classOf[AvroWriteSupport])
ParquetOutputFormat.setCompression(hadoopJob, CompressionCodecName.GZIP)
AvroWriteSupport.setSchema(hadoopJob.getConfiguration, schema)

new PairRDDFunctions[Void,C](
source.map(sourceRecord => (null, sourceRecord))
).saveAsNewAPIHadoopFile(
bucketDAO.uri(dstPath),
classOf[Void], //K
ctag.runtimeClass.asInstanceOf[Class[C]], //V
classOf[AvroParquetOutputFormat],
hadoopJob.getConfiguration
)
}

def readParquet[C](srcPath: String, schema: org.apache.avro.Schema)(implicit ctag: ClassTag[C]): RDD[C] = {
val hadoopJob = Job.getInstance()
ParquetInputFormat.setReadSupportClass(hadoopJob, classOf[AvroReadSupport[C]])
AvroReadSupport.setAvroReadSchema(hadoopJob.getConfiguration, schema)
sc.newAPIHadoopFile(
bucketDAO.uri(srcPath),
classOf[ParquetInputFormat[C]],
classOf[Void], //K
ctag.runtimeClass.asInstanceOf[Class[C]], //V
hadoopJob.getConfiguration
).map { _._2 }
}
```
We use avro-tools to generate java classes from our avro schemas.
java -jar /path/to/avro-tools-1.8.0.jar compile schema

The test case harvests zeroes as values of mandatoryIntField

Naively, I see a problem in the [indexed revord converter](https://git-wip-us.apache.org/repos/asf?p=parquet-mr.git;a=blob;f=parquet-avro/src/main/java/org/apache/parquet/avro/AvroIndexedRecordConverter.java;h=06c66d692571da08298ae1da4f9967446c4864ee;hb=HEAD#l105) in that it cheerfully accepts a condition doomed to fail. The condition being: the reader schema has a column with no default value that is absent in the writer schema.

I am writing predominantly to confirm my diagnosis and to get the intell on why is it implemented the way it is. Is it fixable (or other depend on it as on a feature)? Can people think of a workaround?

**Reporter**: [Remek Zajac](https://issues.apache.org/jira/secure/ViewProfile.jspa?name=remek.zajac@gmail.com)

**Note**: *This issue was originally created as [PARQUET-577](https://issues.apache.org/jira/browse/PARQUET-577). Please see the [migration documentation](https://issues.apache.org/jira/browse/PARQUET-2502) for further details.*

Hướng dẫn đóng góp

Chưa lập chỉ mục được hướng dẫn đóng góp cho kho mã nguồn này

Hướng nghiên cứu

Bắt đầu với phần triển khai AvroIndexedRecordConverter được liên kết và test case MockAvroParquetGrid bằng cách sử dụng các writer schema và reader schema được cung cấp. Tái hiện việc đọc một mandatoryIntField bị thiếu, sau đó theo dõi cách giá trị bị thiếu của nó trở thành zero; hoàn thành có nghĩa là schema không tương thích bị từ chối thay vì âm thầm tạo ra một giá trị.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Đánh giá

Công nghệ
java
Lĩnh vực
data-engineering
Loại issue
Lỗi
Độ khó
3/5
Thời gian dự kiến
1-2 ngày
Mức độ hoạt động
Đình trệ
Độ rõ ràng
Khá rõ ràng
Mức phù hợp với người mới
38/100

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.