Skip to content

[SPARK-2489][SQL] Support Parquet's optional fixed_len_byte_array #35902

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
nicolaslrveiga wants to merge 4 commits into from
Closed

[SPARK-2489][SQL] Support Parquet's optional fixed_len_byte_array #35902

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7a5dd1d
[SPARK-2489][SQL] Support Parquet's optional fixed_len_byte_array
Mar 17, 2022
719cbe9
Merge branch 'apache:master' into SPARK-2489
nicolaslrveiga Mar 20, 2022
a9109d4
Remove new line and fix linter findings
Mar 20, 2022
f079002
Add Unit type to explicitly return type of functions
Mar 20, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion .../java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,8 @@ public ParquetVectorUpdater getUpdater(ColumnDescriptor descriptor, DataType spa
return new FixedLenByteArrayAsIntUpdater(arrayLen);
} else if (canReadAsLongDecimal(descriptor, sparkType)) {
return new FixedLenByteArrayAsLongUpdater(arrayLen);
} else if (canReadAsBinaryDecimal(descriptor, sparkType)) {
} else if (canReadAsBinaryDecimal(descriptor, sparkType) ||
sparkType == DataTypes.BinaryType) {
return new FixedLenByteArrayUpdater(arrayLen);
}
break;
Expand Down
1 change: 1 addition & 0 deletions ...ain/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,6 +274,7 @@ class ParquetToSparkSchemaConverter(
case _: DecimalLogicalTypeAnnotation =>
makeDecimalType(Decimal.maxPrecisionForBytes(parquetType.getTypeLength))
case _: IntervalLogicalTypeAnnotation => typeNotImplemented()
case null => BinaryType
case _ => illegalType()
}

Expand Down
4 changes: 4 additions & 0 deletions sql/core/src/test/avro/parquet-compat.avdl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ protocol CompatibilityTest {
string nested_string_column;
}

fixed FixedType(8);

record AvroPrimitives {
boolean bool_column;
int int_column;
Expand All @@ -42,6 +44,7 @@ protocol CompatibilityTest {
double double_column;
bytes binary_column;
string string_column;
FixedType fixed_column;
}

record AvroOptionalPrimitives {
Expand All @@ -52,6 +55,7 @@ protocol CompatibilityTest {
union { null, double } maybe_double_column;
union { null, bytes } maybe_binary_column;
union { null, string } maybe_string_column;
union { null, FixedType} maybe_fixed_column;
}

record AvroNonNullableArrays {
Expand Down
10 changes: 10 additions & 0 deletions sql/core/src/test/avro/parquet-compat.avpr
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,10 @@
"name" : "nested_string_column",
"type" : "string"
} ]
}, {
"type" : "fixed",
"name" : "FixedType",
"size" : 8
}, {
"type" : "record",
"name" : "AvroPrimitives",
Expand All @@ -49,6 +53,9 @@
}, {
"name" : "string_column",
"type" : "string"
}, {
"name" : "fixed_column",
"type" : "FixedType"
} ]
}, {
"type" : "record",
Expand All @@ -74,6 +81,9 @@
}, {
"name" : "maybe_string_column",
"type" : [ "null", "string" ]
}, {
"name" : "maybe_fixed_column",
"type" : [ "null", "FixedType" ]
} ]
}, {
"type" : "record",
Expand Down
74 changes: 72 additions & 2 deletions .../org/apache/spark/sql/execution/datasources/parquet/test/avro/AvroOptionalPrimitives.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
@SuppressWarnings("all")
@org.apache.avro.specific.AvroGenerated
public class AvroOptionalPrimitives extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"AvroOptionalPrimitives\",\"namespace\":\"org.apache.spark.sql.execution.datasources.parquet.test.avro\",\"fields\":[{\"name\":\"maybe_bool_column\",\"type\":[\"null\",\"boolean\"]},{\"name\":\"maybe_int_column\",\"type\":[\"null\",\"int\"]},{\"name\":\"maybe_long_column\",\"type\":[\"null\",\"long\"]},{\"name\":\"maybe_float_column\",\"type\":[\"null\",\"float\"]},{\"name\":\"maybe_double_column\",\"type\":[\"null\",\"double\"]},{\"name\":\"maybe_binary_column\",\"type\":[\"null\",\"bytes\"]},{\"name\":\"maybe_string_column\",\"type\":[\"null\",{\"type\":\"string\",\"avro.java.string\":\"String\"}]}]}");
public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"AvroOptionalPrimitives\",\"namespace\":\"org.apache.spark.sql.execution.datasources.parquet.test.avro\",\"fields\":[{\"name\":\"maybe_bool_column\",\"type\":[\"null\",\"boolean\"]},{\"name\":\"maybe_int_column\",\"type\":[\"null\",\"int\"]},{\"name\":\"maybe_long_column\",\"type\":[\"null\",\"long\"]},{\"name\":\"maybe_float_column\",\"type\":[\"null\",\"float\"]},{\"name\":\"maybe_double_column\",\"type\":[\"null\",\"double\"]},{\"name\":\"maybe_binary_column\",\"type\":[\"null\",\"bytes\"]},{\"name\":\"maybe_string_column\",\"type\":[\"null\",{\"type\":\"string\",\"avro.java.string\":\"String\"}]},{\"name\":\"maybe_fixed_column\",\"type\":[\"null\",{\"type\":\"fixed\",\"name\":\"FixedType\",\"size\":8}]}]}");
public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
@Deprecated public java.lang.Boolean maybe_bool_column;
@Deprecated public java.lang.Integer maybe_int_column;
Expand All @@ -17,6 +17,8 @@ public class AvroOptionalPrimitives extends org.apache.avro.specific.SpecificRec
@Deprecated public java.nio.ByteBuffer maybe_binary_column;
@Deprecated public java.lang.String maybe_string_column;

private org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType maybe_fixed_column;

/**
* Default constructor. Note that this does not initialize fields
* to their default values from the schema. If that is desired then
Expand All @@ -27,14 +29,15 @@ public AvroOptionalPrimitives() {}
/**
* All-args constructor.
*/
public AvroOptionalPrimitives(java.lang.Boolean maybe_bool_column, java.lang.Integer maybe_int_column, java.lang.Long maybe_long_column, java.lang.Float maybe_float_column, java.lang.Double maybe_double_column, java.nio.ByteBuffer maybe_binary_column, java.lang.String maybe_string_column) {
public AvroOptionalPrimitives(java.lang.Boolean maybe_bool_column, java.lang.Integer maybe_int_column, java.lang.Long maybe_long_column, java.lang.Float maybe_float_column, java.lang.Double maybe_double_column, java.nio.ByteBuffer maybe_binary_column, java.lang.String maybe_string_column, org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType maybe_fixed_column) {
this.maybe_bool_column = maybe_bool_column;
this.maybe_int_column = maybe_int_column;
this.maybe_long_column = maybe_long_column;
this.maybe_float_column = maybe_float_column;
this.maybe_double_column = maybe_double_column;
this.maybe_binary_column = maybe_binary_column;
this.maybe_string_column = maybe_string_column;
this.maybe_fixed_column = maybe_fixed_column;
}

public org.apache.avro.Schema getSchema() { return SCHEMA$; }
Expand All @@ -48,6 +51,7 @@ public java.lang.Object get(int field$) {
case 4: return maybe_double_column;
case 5: return maybe_binary_column;
case 6: return maybe_string_column;
case 7: return maybe_fixed_column;
default: throw new org.apache.avro.AvroRuntimeException("Bad index");
}
}
Expand All @@ -62,6 +66,7 @@ public void put(int field$, java.lang.Object value$) {
case 4: maybe_double_column = (java.lang.Double)value$; break;
case 5: maybe_binary_column = (java.nio.ByteBuffer)value$; break;
case 6: maybe_string_column = (java.lang.String)value$; break;
case 7: maybe_fixed_column = (org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType)value$; break;
default: throw new org.apache.avro.AvroRuntimeException("Bad index");
}
}
Expand Down Expand Up @@ -171,6 +176,22 @@ public void setMaybeStringColumn(java.lang.String value) {
this.maybe_string_column = value;
}

/**
* Gets the value of the 'maybe_fixed_column' field.
* @return The value of the 'maybe_fixed_column' field.
*/
public org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType getMaybeFixedColumn() {
return maybe_fixed_column;
}

/**
* Sets the value of the 'maybe_fixed_column' field.
* @param value the value to set.
*/
public void setMaybeFixedColumn(org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType value) {
this.maybe_fixed_column = value;
}

/** Creates a new AvroOptionalPrimitives RecordBuilder */
public static org.apache.spark.sql.execution.datasources.parquet.test.avro.AvroOptionalPrimitives.Builder newBuilder() {
return new org.apache.spark.sql.execution.datasources.parquet.test.avro.AvroOptionalPrimitives.Builder();
Expand Down Expand Up @@ -199,6 +220,7 @@ public static class Builder extends org.apache.avro.specific.SpecificRecordBuild
private java.lang.Double maybe_double_column;
private java.nio.ByteBuffer maybe_binary_column;
private java.lang.String maybe_string_column;
private org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType maybe_fixed_column;

/** Creates a new Builder */
private Builder() {
Expand Down Expand Up @@ -236,6 +258,10 @@ private Builder(org.apache.spark.sql.execution.datasources.parquet.test.avro.Avr
this.maybe_string_column = data().deepCopy(fields()[6].schema(), other.maybe_string_column);
fieldSetFlags()[6] = true;
}
if (isValidValue(fields()[7], other.maybe_fixed_column)) {
this.maybe_fixed_column = data().deepCopy(fields()[7].schema(), other.maybe_fixed_column);
fieldSetFlags()[7] = true;
}
}

/** Creates a Builder by copying an existing AvroOptionalPrimitives instance */
Expand Down Expand Up @@ -269,6 +295,10 @@ private Builder(org.apache.spark.sql.execution.datasources.parquet.test.avro.Avr
this.maybe_string_column = data().deepCopy(fields()[6].schema(), other.maybe_string_column);
fieldSetFlags()[6] = true;
}
if (isValidValue(fields()[7], other.maybe_fixed_column)) {
this.maybe_fixed_column = data().deepCopy(fields()[7].schema(), other.maybe_fixed_column);
fieldSetFlags()[7] = true;
}
}

/** Gets the value of the 'maybe_bool_column' field */
Expand Down Expand Up @@ -446,6 +476,45 @@ public org.apache.spark.sql.execution.datasources.parquet.test.avro.AvroOptional
return this;
}

/**
* Gets the value of the 'maybe_fixed_column' field.
* @return The value.
*/
public org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType getMaybeFixedColumn() {
return maybe_fixed_column;
}

/**
* Sets the value of the 'maybe_fixed_column' field.
* @param value The value of 'maybe_fixed_column'.
* @return This builder.
*/
public org.apache.spark.sql.execution.datasources.parquet.test.avro.AvroOptionalPrimitives.Builder setMaybeFixedColumn(org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType value) {
validate(fields()[7], value);
this.maybe_fixed_column = value;
fieldSetFlags()[7] = true;
return this;
}

/**
* Checks whether the 'maybe_fixed_column' field has been set.
* @return True if the 'maybe_fixed_column' field has been set, false otherwise.
*/
public boolean hasMaybeFixedColumn() {
return fieldSetFlags()[7];
}


/**
* Clears the value of the 'maybe_fixed_column' field.
* @return This builder.
*/
public org.apache.spark.sql.execution.datasources.parquet.test.avro.AvroOptionalPrimitives.Builder clearMaybeFixedColumn() {
maybe_fixed_column = null;
fieldSetFlags()[7] = false;
return this;
}

@Override
public AvroOptionalPrimitives build() {
try {
Expand All @@ -457,6 +526,7 @@ public AvroOptionalPrimitives build() {
record.maybe_double_column = fieldSetFlags()[4] ? this.maybe_double_column : (java.lang.Double) defaultValue(fields()[4]);
record.maybe_binary_column = fieldSetFlags()[5] ? this.maybe_binary_column : (java.nio.ByteBuffer) defaultValue(fields()[5]);
record.maybe_string_column = fieldSetFlags()[6] ? this.maybe_string_column : (java.lang.String) defaultValue(fields()[6]);
record.maybe_fixed_column = fieldSetFlags()[7] ? this.maybe_fixed_column : (org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType) defaultValue(fields()[7]);
return record;
} catch (Exception e) {
throw new org.apache.avro.AvroRuntimeException(e);
Expand Down
72 changes: 70 additions & 2 deletions ...gen-java/org/apache/spark/sql/execution/datasources/parquet/test/avro/AvroPrimitives.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
@SuppressWarnings("all")
@org.apache.avro.specific.AvroGenerated
public class AvroPrimitives extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"AvroPrimitives\",\"namespace\":\"org.apache.spark.sql.execution.datasources.parquet.test.avro\",\"fields\":[{\"name\":\"bool_column\",\"type\":\"boolean\"},{\"name\":\"int_column\",\"type\":\"int\"},{\"name\":\"long_column\",\"type\":\"long\"},{\"name\":\"float_column\",\"type\":\"float\"},{\"name\":\"double_column\",\"type\":\"double\"},{\"name\":\"binary_column\",\"type\":\"bytes\"},{\"name\":\"string_column\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}]}");
public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"AvroPrimitives\",\"namespace\":\"org.apache.spark.sql.execution.datasources.parquet.test.avro\",\"fields\":[{\"name\":\"bool_column\",\"type\":\"boolean\"},{\"name\":\"int_column\",\"type\":\"int\"},{\"name\":\"long_column\",\"type\":\"long\"},{\"name\":\"float_column\",\"type\":\"float\"},{\"name\":\"double_column\",\"type\":\"double\"},{\"name\":\"binary_column\",\"type\":\"bytes\"},{\"name\":\"string_column\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}},{\"name\":\"fixed_column\",\"type\":{\"type\":\"fixed\",\"name\":\"FixedType\",\"size\":8}}]}");
public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
@Deprecated public boolean bool_column;
@Deprecated public int int_column;
Expand All @@ -17,6 +17,8 @@ public class AvroPrimitives extends org.apache.avro.specific.SpecificRecordBase
@Deprecated public java.nio.ByteBuffer binary_column;
@Deprecated public java.lang.String string_column;

private org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType fixed_column;

/**
* Default constructor. Note that this does not initialize fields
* to their default values from the schema. If that is desired then
Expand All @@ -27,14 +29,15 @@ public AvroPrimitives() {}
/**
* All-args constructor.
*/
public AvroPrimitives(java.lang.Boolean bool_column, java.lang.Integer int_column, java.lang.Long long_column, java.lang.Float float_column, java.lang.Double double_column, java.nio.ByteBuffer binary_column, java.lang.String string_column) {
public AvroPrimitives(java.lang.Boolean bool_column, java.lang.Integer int_column, java.lang.Long long_column, java.lang.Float float_column, java.lang.Double double_column, java.nio.ByteBuffer binary_column, java.lang.String string_column, org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType fixed_column) {
this.bool_column = bool_column;
this.int_column = int_column;
this.long_column = long_column;
this.float_column = float_column;
this.double_column = double_column;
this.binary_column = binary_column;
this.string_column = string_column;
this.fixed_column = fixed_column;
}

public org.apache.avro.Schema getSchema() { return SCHEMA$; }
Expand All @@ -48,6 +51,7 @@ public java.lang.Object get(int field$) {
case 4: return double_column;
case 5: return binary_column;
case 6: return string_column;
case 7: return fixed_column;
default: throw new org.apache.avro.AvroRuntimeException("Bad index");
}
}
Expand All @@ -62,6 +66,7 @@ public void put(int field$, java.lang.Object value$) {
case 4: double_column = (java.lang.Double)value$; break;
case 5: binary_column = (java.nio.ByteBuffer)value$; break;
case 6: string_column = (java.lang.String)value$; break;
case 7: fixed_column = (org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType)value$; break;
default: throw new org.apache.avro.AvroRuntimeException("Bad index");
}
}
Expand Down Expand Up @@ -171,6 +176,21 @@ public void setStringColumn(java.lang.String value) {
this.string_column = value;
}

/**
* Gets the value of the 'fixed_column' field.
*/
public org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType getFixedColumn() {
return fixed_column;
}

/**
* Sets the value of the 'fixed_column' field.
* @param value the value to set.
*/
public void setFixedColumn(org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType value) {
this.fixed_column = value;
}

/** Creates a new AvroPrimitives RecordBuilder */
public static org.apache.spark.sql.execution.datasources.parquet.test.avro.AvroPrimitives.Builder newBuilder() {
return new org.apache.spark.sql.execution.datasources.parquet.test.avro.AvroPrimitives.Builder();
Expand Down Expand Up @@ -199,6 +219,7 @@ public static class Builder extends org.apache.avro.specific.SpecificRecordBuild
private double double_column;
private java.nio.ByteBuffer binary_column;
private java.lang.String string_column;
private org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType fixed_column;

/** Creates a new Builder */
private Builder() {
Expand Down Expand Up @@ -236,6 +257,10 @@ private Builder(org.apache.spark.sql.execution.datasources.parquet.test.avro.Avr
this.string_column = data().deepCopy(fields()[6].schema(), other.string_column);
fieldSetFlags()[6] = true;
}
if (isValidValue(fields()[7], other.fixed_column)) {
this.fixed_column = data().deepCopy(fields()[7].schema(), other.fixed_column);
fieldSetFlags()[7] = true;
}
}

/** Creates a Builder by copying an existing AvroPrimitives instance */
Expand Down Expand Up @@ -269,6 +294,10 @@ private Builder(org.apache.spark.sql.execution.datasources.parquet.test.avro.Avr
this.string_column = data().deepCopy(fields()[6].schema(), other.string_column);
fieldSetFlags()[6] = true;
}
if (isValidValue(fields()[7], other.fixed_column)) {
this.fixed_column = data().deepCopy(fields()[7].schema(), other.fixed_column);
fieldSetFlags()[7] = true;
}
}

/** Gets the value of the 'bool_column' field */
Expand Down Expand Up @@ -441,6 +470,44 @@ public org.apache.spark.sql.execution.datasources.parquet.test.avro.AvroPrimitiv
return this;
}

/**
* Gets the value of the 'fixed_column' field.
* @return The value.
*/
public org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType getFixedColumn() {
return fixed_column;
}

/**
* Sets the value of the 'fixed_column' field.
* @param value The value of 'fixed_column'.
* @return This builder.
*/
public org.apache.spark.sql.execution.datasources.parquet.test.avro.AvroPrimitives.Builder setFixedColumn(org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType value) {
validate(fields()[7], value);
this.fixed_column = value;
fieldSetFlags()[7] = true;
return this;
}

/**
* Checks whether the 'fixed_column' field has been set.
* @return True if the 'fixed_column' field has been set, false otherwise.
*/
public boolean hasFixedColumn() {
return fieldSetFlags()[7];
}

/**
* Clears the value of the 'fixed_column' field.
* @return This builder.
*/
public org.apache.spark.sql.execution.datasources.parquet.test.avro.AvroPrimitives.Builder clearFixedColumn() {
fixed_column = null;
fieldSetFlags()[7] = false;
return this;
}

@Override
public AvroPrimitives build() {
try {
Expand All @@ -452,6 +519,7 @@ public AvroPrimitives build() {
record.double_column = fieldSetFlags()[4] ? this.double_column : (java.lang.Double) defaultValue(fields()[4]);
record.binary_column = fieldSetFlags()[5] ? this.binary_column : (java.nio.ByteBuffer) defaultValue(fields()[5]);
record.string_column = fieldSetFlags()[6] ? this.string_column : (java.lang.String) defaultValue(fields()[6]);
record.fixed_column = fieldSetFlags()[7] ? this.fixed_column : (org.apache.spark.sql.execution.datasources.parquet.test.avro.FixedType) defaultValue(fields()[7]);
return record;
} catch (Exception e) {
throw new org.apache.avro.AvroRuntimeException(e);
Expand Down
Loading