You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2021/10/08 09:53:40 UTC
[GitHub] [iceberg] szlta opened a new pull request #3248: Refactor ORC value writers
szlta opened a new pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248
Currently there are 3 struct-like data holder types that ORC integrates with in Iceberg: Record (Iceberg), RowData (Flink) and InternalRow (Spark).
While there's an OrcRowWriter<T> implementation for all these 3 types, only the first two relies on OrcValueWriter<?>'s, while Spark uses SparkValueWriters and that has a different API ("int ordinal" is part of nonNullWrite method).
Also there's no common abstract StuctWriter<?> class which would be similar to how Parquet/Avro writes are implemented (e.g. see https://github.com/apache/iceberg/blob/master/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java#L539 )
Due to all that, the current ORC write code has a lot of code duplication, and fits less into Iceberg, where the write code paths for Parquet and Avro are much more in harmony.
Thus I propose to bring ORC writes up to the standard too, as we can already see some effects of this shortcoming in this PR: https://github.com/apache/iceberg/pull/2935 where a common StructWriter<?> type could help implement positional delete writer creation for ORC.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r725845536
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriter.java
##########
@@ -115,53 +116,72 @@ private WriteBuilder() {
}
@Override
- @SuppressWarnings("unchecked")
public void write(Record value, VectorizedRowBatch output) {
- Preconditions.checkArgument(writer instanceof RecordWriter, "writer must be a RecordWriter.");
+ Preconditions.checkArgument(value != null, "value must not be null");
int row = output.size;
output.size += 1;
- List<OrcValueWriter<?>> writers = ((RecordWriter) writer).writers();
- for (int c = 0; c < writers.size(); ++c) {
- OrcValueWriter child = writers.get(c);
- child.write(row, value.get(c, child.getJavaClass()), output.cols[c]);
- }
+
+ writer.rootNonNullWrite(row, value, output);
}
@Override
public Stream<FieldMetrics<?>> metrics() {
return writer.metrics();
}
- private static class RecordWriter implements OrcValueWriter<Record> {
+ public abstract static class StructWriter<S> implements OrcValueWriter<S> {
private final List<OrcValueWriter<?>> writers;
- RecordWriter(List<OrcValueWriter<?>> writers) {
+ protected StructWriter(List<OrcValueWriter<?>> writers) {
this.writers = writers;
}
- List<OrcValueWriter<?>> writers() {
- return writers;
+ public List<OrcValueWriter<?>> writers() {
+ return this.writers;
}
@Override
- public Class<Record> getJavaClass() {
- return Record.class;
+ public Stream<FieldMetrics<?>> metrics() {
+ return writers.stream().flatMap(OrcValueWriter::metrics);
}
@Override
- @SuppressWarnings("unchecked")
- public void nonNullWrite(int rowId, Record data, ColumnVector output) {
+ public void nonNullWrite(int rowId, S value, ColumnVector output) {
StructColumnVector cv = (StructColumnVector) output;
+ write(rowId, value, c -> cv.fields[c]);
+ }
+
+ // Special case of writing the root struct
+ public void rootNonNullWrite(int rowId, S value, VectorizedRowBatch output) {
+ write(rowId, value, c -> output.cols[c]);
+ }
+
+ private void write(int rowId, S value, Function<Integer, ColumnVector> colVectorAtFunc) {
for (int c = 0; c < writers.size(); ++c) {
- OrcValueWriter child = writers.get(c);
- child.write(rowId, data.get(c, child.getJavaClass()), cv.fields[c]);
+ OrcValueWriter writer = writers.get(c);
+ writer.write(rowId, get(value, c), colVectorAtFunc.apply(c));
}
}
+ protected abstract Object get(S struct, int index);
+ }
+
+ private static class RecordWriter extends StructWriter<Record> {
+
+ RecordWriter(List<OrcValueWriter<?>> writers) {
+ super(writers);
+ }
+
@Override
- public Stream<FieldMetrics<?>> metrics() {
- return writers.stream().flatMap(OrcValueWriter::metrics);
+ public Class<Record> getJavaClass() {
+ return Record.class;
+ }
+
+ @Override
+ protected Object get(Record struct, int index) {
+ return struct.get(index, writers().get(index).getJavaClass());
}
+
Review comment:
Nit: extra empty line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szlta commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
szlta commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r727880566
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -19,245 +19,130 @@
package org.apache.iceberg.spark.data;
+import java.util.List;
import java.util.stream.Stream;
-import org.apache.iceberg.DoubleFieldMetrics;
import org.apache.iceberg.FieldMetrics;
-import org.apache.iceberg.FloatFieldMetrics;
+import org.apache.iceberg.orc.OrcValueWriter;
+import org.apache.orc.TypeDescription;
import org.apache.orc.storage.common.type.HiveDecimal;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
-import org.apache.orc.storage.ql.exec.vector.DoubleColumnVector;
import org.apache.orc.storage.ql.exec.vector.ListColumnVector;
-import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
import org.apache.orc.storage.ql.exec.vector.MapColumnVector;
import org.apache.orc.storage.ql.exec.vector.TimestampColumnVector;
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
import org.apache.spark.sql.catalyst.util.ArrayData;
import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
class SparkOrcValueWriters {
private SparkOrcValueWriters() {
}
- static SparkOrcValueWriter booleans() {
- return BooleanWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter bytes() {
- return ByteWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter shorts() {
- return ShortWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter ints() {
- return IntWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter longs() {
- return LongWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter floats(int id) {
- return new FloatWriter(id);
- }
-
- static SparkOrcValueWriter doubles(int id) {
- return new DoubleWriter(id);
- }
-
- static SparkOrcValueWriter byteArrays() {
- return BytesWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter strings() {
+ static OrcValueWriter<?> strings() {
return StringWriter.INSTANCE;
}
- static SparkOrcValueWriter timestampTz() {
+ static OrcValueWriter<?> timestampTz() {
return TimestampTzWriter.INSTANCE;
}
- static SparkOrcValueWriter decimal(int precision, int scale) {
+ static OrcValueWriter<?> decimal(int precision, int scale) {
if (precision <= 18) {
- return new Decimal18Writer(precision, scale);
+ return new Decimal18Writer(scale);
} else {
- return new Decimal38Writer(precision, scale);
- }
- }
-
- static SparkOrcValueWriter list(SparkOrcValueWriter element) {
- return new ListWriter(element);
- }
-
- static SparkOrcValueWriter map(SparkOrcValueWriter keyWriter, SparkOrcValueWriter valueWriter) {
- return new MapWriter(keyWriter, valueWriter);
- }
-
- private static class BooleanWriter implements SparkOrcValueWriter {
- private static final BooleanWriter INSTANCE = new BooleanWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getBoolean(column) ? 1 : 0;
- }
- }
-
- private static class ByteWriter implements SparkOrcValueWriter {
- private static final ByteWriter INSTANCE = new ByteWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getByte(column);
- }
- }
-
- private static class ShortWriter implements SparkOrcValueWriter {
- private static final ShortWriter INSTANCE = new ShortWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getShort(column);
+ return new Decimal38Writer();
}
}
- private static class IntWriter implements SparkOrcValueWriter {
- private static final IntWriter INSTANCE = new IntWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getInt(column);
- }
+ static OrcValueWriter<?> list(OrcValueWriter<?> element, List<TypeDescription> orcType) {
+ return new ListWriter(element, orcType);
}
- private static class LongWriter implements SparkOrcValueWriter {
- private static final LongWriter INSTANCE = new LongWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getLong(column);
- }
+ static OrcValueWriter<?> map(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter,
+ List<TypeDescription> orcType) {
+ return new MapWriter(keyWriter, valueWriter, orcType);
}
- private static class FloatWriter implements SparkOrcValueWriter {
- private final FloatFieldMetrics.Builder floatFieldMetricsBuilder;
-
- private FloatWriter(int id) {
- this.floatFieldMetricsBuilder = new FloatFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- float floatValue = data.getFloat(column);
- ((DoubleColumnVector) output).vector[rowId] = floatValue;
- floatFieldMetricsBuilder.addValue(floatValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(floatFieldMetricsBuilder.build());
- }
- }
-
- private static class DoubleWriter implements SparkOrcValueWriter {
- private final DoubleFieldMetrics.Builder doubleFieldMetricsBuilder;
-
- private DoubleWriter(int id) {
- this.doubleFieldMetricsBuilder = new DoubleFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- double doubleValue = data.getDouble(column);
- ((DoubleColumnVector) output).vector[rowId] = doubleValue;
- doubleFieldMetricsBuilder.addValue(doubleValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(doubleFieldMetricsBuilder.build());
- }
- }
-
- private static class BytesWriter implements SparkOrcValueWriter {
- private static final BytesWriter INSTANCE = new BytesWriter();
+ private static class StringWriter implements OrcValueWriter<UTF8String> {
+ private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- // getBinary always makes a copy, so we don't need to worry about it
- // being changed behind our back.
- byte[] value = data.getBinary(column);
+ public void nonNullWrite(int rowId, UTF8String data, ColumnVector output) {
+ byte[] value = data.getBytes();
((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
}
- }
-
- private static class StringWriter implements SparkOrcValueWriter {
- private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- byte[] value = data.getUTF8String(column).getBytes();
- ((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
+ public Class<?> getJavaClass() {
+ return UTF8String.class;
}
}
- private static class TimestampTzWriter implements SparkOrcValueWriter {
+ private static class TimestampTzWriter implements OrcValueWriter<Long> {
private static final TimestampTzWriter INSTANCE = new TimestampTzWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Long micros, ColumnVector output) {
TimestampColumnVector cv = (TimestampColumnVector) output;
- long micros = data.getLong(column); // it could be negative.
cv.time[rowId] = Math.floorDiv(micros, 1_000); // millis
cv.nanos[rowId] = (int) Math.floorMod(micros, 1_000_000) * 1_000; // nanos
}
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Long.class;
+ }
}
- private static class Decimal18Writer implements SparkOrcValueWriter {
- private final int precision;
+ private static class Decimal18Writer implements OrcValueWriter<Decimal> {
private final int scale;
- Decimal18Writer(int precision, int scale) {
- this.precision = precision;
+ Decimal18Writer(int scale) {
this.scale = scale;
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
((DecimalColumnVector) output).vector[rowId].setFromLongAndScale(
- data.getDecimal(column, precision, scale).toUnscaledLong(), scale);
+ decimal.toUnscaledLong(), scale);
+ }
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class Decimal38Writer implements SparkOrcValueWriter {
- private final int precision;
- private final int scale;
+ private static class Decimal38Writer implements OrcValueWriter<Decimal> {
- Decimal38Writer(int precision, int scale) {
- this.precision = precision;
- this.scale = scale;
+ @Override
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
+ ((DecimalColumnVector) output).vector[rowId].set(
+ HiveDecimal.create(decimal.toJavaBigDecimal()));
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((DecimalColumnVector) output).vector[rowId].set(
- HiveDecimal.create(data.getDecimal(column, precision, scale)
- .toJavaBigDecimal()));
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class ListWriter implements SparkOrcValueWriter {
- private final SparkOrcValueWriter writer;
+ private static class ListWriter implements OrcValueWriter<ArrayData> {
+ private final OrcValueWriter writer;
+ private final SparkOrcWriter.FieldGetter fieldGetter;
Review comment:
Sounds good! Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r725843690
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriter.java
##########
@@ -115,53 +116,72 @@ private WriteBuilder() {
}
@Override
- @SuppressWarnings("unchecked")
public void write(Record value, VectorizedRowBatch output) {
- Preconditions.checkArgument(writer instanceof RecordWriter, "writer must be a RecordWriter.");
+ Preconditions.checkArgument(value != null, "value must not be null");
int row = output.size;
output.size += 1;
- List<OrcValueWriter<?>> writers = ((RecordWriter) writer).writers();
- for (int c = 0; c < writers.size(); ++c) {
- OrcValueWriter child = writers.get(c);
- child.write(row, value.get(c, child.getJavaClass()), output.cols[c]);
- }
+
+ writer.rootNonNullWrite(row, value, output);
}
@Override
public Stream<FieldMetrics<?>> metrics() {
return writer.metrics();
}
- private static class RecordWriter implements OrcValueWriter<Record> {
+ public abstract static class StructWriter<S> implements OrcValueWriter<S> {
private final List<OrcValueWriter<?>> writers;
- RecordWriter(List<OrcValueWriter<?>> writers) {
+ protected StructWriter(List<OrcValueWriter<?>> writers) {
this.writers = writers;
}
- List<OrcValueWriter<?>> writers() {
- return writers;
+ public List<OrcValueWriter<?>> writers() {
+ return this.writers;
Review comment:
Nit: we usually use `this` to assign the value to a class member for distinguishing the variable is a local one or class member in apache iceberg. For this case, we don't need the extra `this.` because it must be a class member.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r726969843
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcWriter.java
##########
@@ -126,30 +122,106 @@ public SparkOrcValueWriter primitive(Type.PrimitiveType iPrimitive, TypeDescript
}
}
- private static class StructWriter implements SparkOrcValueWriter {
- private final List<SparkOrcValueWriter> writers;
+ private static class InternalRowWriter extends GenericOrcWriters.StructWriter<InternalRow> {
+ private final List<FieldGetter<?>> fieldGetters;
- StructWriter(List<SparkOrcValueWriter> writers) {
- this.writers = writers;
- }
+ InternalRowWriter(List<OrcValueWriter<?>> writers, List<TypeDescription> orcTypes) {
+ super(writers);
+ this.fieldGetters = Lists.newArrayListWithExpectedSize(orcTypes.size());
- List<SparkOrcValueWriter> writers() {
- return writers;
+ for (int i = 0; i < orcTypes.size(); i++) {
+ fieldGetters.add(createFieldGetter(orcTypes.get(i)));
+ }
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- InternalRow value = data.getStruct(column, writers.size());
- StructColumnVector cv = (StructColumnVector) output;
- for (int c = 0; c < writers.size(); ++c) {
- writers.get(c).write(rowId, c, value, cv.fields[c]);
- }
+ public Class<?> getJavaClass() {
+ return InternalRow.class;
}
@Override
public Stream<FieldMetrics<?>> metrics() {
- return writers.stream().flatMap(SparkOrcValueWriter::metrics);
+ return writers().stream().flatMap(OrcValueWriter::metrics);
Review comment:
Nit: No need to rewrite this method because seems like it has the same implementation as the super class 's `metrics`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r725837974
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriter.java
##########
@@ -115,53 +116,72 @@ private WriteBuilder() {
}
@Override
- @SuppressWarnings("unchecked")
public void write(Record value, VectorizedRowBatch output) {
- Preconditions.checkArgument(writer instanceof RecordWriter, "writer must be a RecordWriter.");
+ Preconditions.checkArgument(value != null, "value must not be null");
int row = output.size;
output.size += 1;
- List<OrcValueWriter<?>> writers = ((RecordWriter) writer).writers();
- for (int c = 0; c < writers.size(); ++c) {
- OrcValueWriter child = writers.get(c);
- child.write(row, value.get(c, child.getJavaClass()), output.cols[c]);
- }
+
+ writer.rootNonNullWrite(row, value, output);
Review comment:
Copied the comment from here: https://github.com/apache/iceberg/pull/3250#discussion_r725771333
> This generic StructWriter doesn't belongs to the concrete OrcRowWriter<Record>, moving this to the writer definition class GenericOrcWriters looks more reasonable to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r726920053
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriters.java
##########
@@ -531,6 +534,46 @@ public void nonNullWrite(int rowId, Map<K, V> map, ColumnVector output) {
}
}
+ public abstract static class StructWriter<S> implements OrcValueWriter<S> {
+ private final List<OrcValueWriter<?>> writers;
+
+ protected StructWriter(List<OrcValueWriter<?>> writers) {
+ this.writers = writers;
+ }
+
+ public List<OrcValueWriter<?>> writers() {
+ return writers;
+ }
+
+ @Override
+ public Stream<FieldMetrics<?>> metrics() {
+ return writers.stream().flatMap(OrcValueWriter::metrics);
+ }
+
+ @Override
+ public void nonNullWrite(int rowId, S value, ColumnVector output) {
+ StructColumnVector cv = (StructColumnVector) output;
+ write(rowId, value, c -> cv.fields[c]);
+ }
+
+ // Special case of writing the root struct
+ public void rootNonNullWrite(S value, VectorizedRowBatch output) {
Review comment:
Is it better to rename this name to `writeRow` or `addRow` because it is adding a new row ( which is `value`) into the VectorizedRowBatch array. It's no quite clear for me to understand the `rootNonNullWrites` from my first glance actually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r726912272
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriter.java
##########
@@ -150,18 +135,8 @@ public void write(Record value, VectorizedRowBatch output) {
}
@Override
- @SuppressWarnings("unchecked")
- public void nonNullWrite(int rowId, Record data, ColumnVector output) {
- StructColumnVector cv = (StructColumnVector) output;
- for (int c = 0; c < writers.size(); ++c) {
- OrcValueWriter child = writers.get(c);
- child.write(rowId, data.get(c, child.getJavaClass()), cv.fields[c]);
- }
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return writers.stream().flatMap(OrcValueWriter::metrics);
+ protected Object get(Record struct, int index) {
+ return struct.get(index, writers().get(index).getJavaClass());
Review comment:
I think the `Class<?> getJavaClass()` from `OrcValueWriter` can be removed now ha ? Because it's the only place that we will use the `getJavaClass` , while the `struct` is actually a `Record` which means we could just use the `struct#get(index)` here, right ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r725838797
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriter.java
##########
@@ -115,53 +116,72 @@ private WriteBuilder() {
}
@Override
- @SuppressWarnings("unchecked")
public void write(Record value, VectorizedRowBatch output) {
- Preconditions.checkArgument(writer instanceof RecordWriter, "writer must be a RecordWriter.");
+ Preconditions.checkArgument(value != null, "value must not be null");
int row = output.size;
output.size += 1;
- List<OrcValueWriter<?>> writers = ((RecordWriter) writer).writers();
- for (int c = 0; c < writers.size(); ++c) {
- OrcValueWriter child = writers.get(c);
- child.write(row, value.get(c, child.getJavaClass()), output.cols[c]);
- }
+
+ writer.rootNonNullWrite(row, value, output);
}
@Override
public Stream<FieldMetrics<?>> metrics() {
return writer.metrics();
}
- private static class RecordWriter implements OrcValueWriter<Record> {
+ public abstract static class StructWriter<S> implements OrcValueWriter<S> {
private final List<OrcValueWriter<?>> writers;
- RecordWriter(List<OrcValueWriter<?>> writers) {
+ protected StructWriter(List<OrcValueWriter<?>> writers) {
this.writers = writers;
}
- List<OrcValueWriter<?>> writers() {
- return writers;
+ public List<OrcValueWriter<?>> writers() {
+ return this.writers;
}
@Override
- public Class<Record> getJavaClass() {
- return Record.class;
+ public Stream<FieldMetrics<?>> metrics() {
+ return writers.stream().flatMap(OrcValueWriter::metrics);
}
@Override
- @SuppressWarnings("unchecked")
- public void nonNullWrite(int rowId, Record data, ColumnVector output) {
+ public void nonNullWrite(int rowId, S value, ColumnVector output) {
StructColumnVector cv = (StructColumnVector) output;
+ write(rowId, value, c -> cv.fields[c]);
+ }
+
+ // Special case of writing the root struct
+ public void rootNonNullWrite(int rowId, S value, VectorizedRowBatch output) {
Review comment:
Copied the comment from here: https://github.com/apache/iceberg/pull/3250#discussion_r725779961
Why do we need to expose the `rowId` in this `rootNonNullWrite` method ? The root `OrcRowWriter` always provide a write method like :
```java
public void write(RowData row, VectorizedRowBatch output)
```
So why not just hide the rowId inside the `rootNonNullWrite` like:
```java
public void write(RowData row, VectorizedRowBatch output){
Preconditions.checkArgument(row != null, "value must not be null");
writer.rootNonNullWrite(row, output);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szlta commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
szlta commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r727872599
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -19,245 +19,130 @@
package org.apache.iceberg.spark.data;
+import java.util.List;
import java.util.stream.Stream;
-import org.apache.iceberg.DoubleFieldMetrics;
import org.apache.iceberg.FieldMetrics;
-import org.apache.iceberg.FloatFieldMetrics;
+import org.apache.iceberg.orc.OrcValueWriter;
+import org.apache.orc.TypeDescription;
import org.apache.orc.storage.common.type.HiveDecimal;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
-import org.apache.orc.storage.ql.exec.vector.DoubleColumnVector;
import org.apache.orc.storage.ql.exec.vector.ListColumnVector;
-import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
import org.apache.orc.storage.ql.exec.vector.MapColumnVector;
import org.apache.orc.storage.ql.exec.vector.TimestampColumnVector;
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
import org.apache.spark.sql.catalyst.util.ArrayData;
import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
class SparkOrcValueWriters {
private SparkOrcValueWriters() {
}
- static SparkOrcValueWriter booleans() {
- return BooleanWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter bytes() {
- return ByteWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter shorts() {
- return ShortWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter ints() {
- return IntWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter longs() {
- return LongWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter floats(int id) {
- return new FloatWriter(id);
- }
-
- static SparkOrcValueWriter doubles(int id) {
- return new DoubleWriter(id);
- }
-
- static SparkOrcValueWriter byteArrays() {
- return BytesWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter strings() {
+ static OrcValueWriter<?> strings() {
return StringWriter.INSTANCE;
}
- static SparkOrcValueWriter timestampTz() {
+ static OrcValueWriter<?> timestampTz() {
return TimestampTzWriter.INSTANCE;
}
- static SparkOrcValueWriter decimal(int precision, int scale) {
+ static OrcValueWriter<?> decimal(int precision, int scale) {
if (precision <= 18) {
- return new Decimal18Writer(precision, scale);
+ return new Decimal18Writer(scale);
} else {
- return new Decimal38Writer(precision, scale);
- }
- }
-
- static SparkOrcValueWriter list(SparkOrcValueWriter element) {
- return new ListWriter(element);
- }
-
- static SparkOrcValueWriter map(SparkOrcValueWriter keyWriter, SparkOrcValueWriter valueWriter) {
- return new MapWriter(keyWriter, valueWriter);
- }
-
- private static class BooleanWriter implements SparkOrcValueWriter {
- private static final BooleanWriter INSTANCE = new BooleanWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getBoolean(column) ? 1 : 0;
- }
- }
-
- private static class ByteWriter implements SparkOrcValueWriter {
- private static final ByteWriter INSTANCE = new ByteWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getByte(column);
- }
- }
-
- private static class ShortWriter implements SparkOrcValueWriter {
- private static final ShortWriter INSTANCE = new ShortWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getShort(column);
+ return new Decimal38Writer();
}
}
- private static class IntWriter implements SparkOrcValueWriter {
- private static final IntWriter INSTANCE = new IntWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getInt(column);
- }
+ static OrcValueWriter<?> list(OrcValueWriter<?> element, List<TypeDescription> orcType) {
+ return new ListWriter(element, orcType);
}
- private static class LongWriter implements SparkOrcValueWriter {
- private static final LongWriter INSTANCE = new LongWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getLong(column);
- }
+ static OrcValueWriter<?> map(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter,
+ List<TypeDescription> orcType) {
Review comment:
It actually should be named orcTypes - I'll correct that in the next round.
It is required so that we can determine the two fieldGetter types that will be used for retrieving the keys and values of the MapData object. This works similarly to the ListWriter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r727064468
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -19,245 +19,130 @@
package org.apache.iceberg.spark.data;
+import java.util.List;
import java.util.stream.Stream;
-import org.apache.iceberg.DoubleFieldMetrics;
import org.apache.iceberg.FieldMetrics;
-import org.apache.iceberg.FloatFieldMetrics;
+import org.apache.iceberg.orc.OrcValueWriter;
+import org.apache.orc.TypeDescription;
import org.apache.orc.storage.common.type.HiveDecimal;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
-import org.apache.orc.storage.ql.exec.vector.DoubleColumnVector;
import org.apache.orc.storage.ql.exec.vector.ListColumnVector;
-import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
import org.apache.orc.storage.ql.exec.vector.MapColumnVector;
import org.apache.orc.storage.ql.exec.vector.TimestampColumnVector;
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
import org.apache.spark.sql.catalyst.util.ArrayData;
import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
class SparkOrcValueWriters {
private SparkOrcValueWriters() {
}
- static SparkOrcValueWriter booleans() {
- return BooleanWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter bytes() {
- return ByteWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter shorts() {
- return ShortWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter ints() {
- return IntWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter longs() {
- return LongWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter floats(int id) {
- return new FloatWriter(id);
- }
-
- static SparkOrcValueWriter doubles(int id) {
- return new DoubleWriter(id);
- }
-
- static SparkOrcValueWriter byteArrays() {
- return BytesWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter strings() {
+ static OrcValueWriter<?> strings() {
return StringWriter.INSTANCE;
}
- static SparkOrcValueWriter timestampTz() {
+ static OrcValueWriter<?> timestampTz() {
return TimestampTzWriter.INSTANCE;
}
- static SparkOrcValueWriter decimal(int precision, int scale) {
+ static OrcValueWriter<?> decimal(int precision, int scale) {
if (precision <= 18) {
- return new Decimal18Writer(precision, scale);
+ return new Decimal18Writer(scale);
} else {
- return new Decimal38Writer(precision, scale);
- }
- }
-
- static SparkOrcValueWriter list(SparkOrcValueWriter element) {
- return new ListWriter(element);
- }
-
- static SparkOrcValueWriter map(SparkOrcValueWriter keyWriter, SparkOrcValueWriter valueWriter) {
- return new MapWriter(keyWriter, valueWriter);
- }
-
- private static class BooleanWriter implements SparkOrcValueWriter {
- private static final BooleanWriter INSTANCE = new BooleanWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getBoolean(column) ? 1 : 0;
- }
- }
-
- private static class ByteWriter implements SparkOrcValueWriter {
- private static final ByteWriter INSTANCE = new ByteWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getByte(column);
- }
- }
-
- private static class ShortWriter implements SparkOrcValueWriter {
- private static final ShortWriter INSTANCE = new ShortWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getShort(column);
+ return new Decimal38Writer();
}
}
- private static class IntWriter implements SparkOrcValueWriter {
- private static final IntWriter INSTANCE = new IntWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getInt(column);
- }
+ static OrcValueWriter<?> list(OrcValueWriter<?> element, List<TypeDescription> orcType) {
+ return new ListWriter(element, orcType);
}
- private static class LongWriter implements SparkOrcValueWriter {
- private static final LongWriter INSTANCE = new LongWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getLong(column);
- }
+ static OrcValueWriter<?> map(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter,
+ List<TypeDescription> orcType) {
+ return new MapWriter(keyWriter, valueWriter, orcType);
}
- private static class FloatWriter implements SparkOrcValueWriter {
- private final FloatFieldMetrics.Builder floatFieldMetricsBuilder;
-
- private FloatWriter(int id) {
- this.floatFieldMetricsBuilder = new FloatFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- float floatValue = data.getFloat(column);
- ((DoubleColumnVector) output).vector[rowId] = floatValue;
- floatFieldMetricsBuilder.addValue(floatValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(floatFieldMetricsBuilder.build());
- }
- }
-
- private static class DoubleWriter implements SparkOrcValueWriter {
- private final DoubleFieldMetrics.Builder doubleFieldMetricsBuilder;
-
- private DoubleWriter(int id) {
- this.doubleFieldMetricsBuilder = new DoubleFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- double doubleValue = data.getDouble(column);
- ((DoubleColumnVector) output).vector[rowId] = doubleValue;
- doubleFieldMetricsBuilder.addValue(doubleValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(doubleFieldMetricsBuilder.build());
- }
- }
-
- private static class BytesWriter implements SparkOrcValueWriter {
- private static final BytesWriter INSTANCE = new BytesWriter();
+ private static class StringWriter implements OrcValueWriter<UTF8String> {
+ private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- // getBinary always makes a copy, so we don't need to worry about it
- // being changed behind our back.
- byte[] value = data.getBinary(column);
+ public void nonNullWrite(int rowId, UTF8String data, ColumnVector output) {
+ byte[] value = data.getBytes();
((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
}
- }
-
- private static class StringWriter implements SparkOrcValueWriter {
- private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- byte[] value = data.getUTF8String(column).getBytes();
- ((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
+ public Class<?> getJavaClass() {
+ return UTF8String.class;
}
}
- private static class TimestampTzWriter implements SparkOrcValueWriter {
+ private static class TimestampTzWriter implements OrcValueWriter<Long> {
private static final TimestampTzWriter INSTANCE = new TimestampTzWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Long micros, ColumnVector output) {
TimestampColumnVector cv = (TimestampColumnVector) output;
- long micros = data.getLong(column); // it could be negative.
cv.time[rowId] = Math.floorDiv(micros, 1_000); // millis
cv.nanos[rowId] = (int) Math.floorMod(micros, 1_000_000) * 1_000; // nanos
}
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Long.class;
+ }
}
- private static class Decimal18Writer implements SparkOrcValueWriter {
- private final int precision;
+ private static class Decimal18Writer implements OrcValueWriter<Decimal> {
private final int scale;
- Decimal18Writer(int precision, int scale) {
- this.precision = precision;
+ Decimal18Writer(int scale) {
this.scale = scale;
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
((DecimalColumnVector) output).vector[rowId].setFromLongAndScale(
- data.getDecimal(column, precision, scale).toUnscaledLong(), scale);
+ decimal.toUnscaledLong(), scale);
+ }
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class Decimal38Writer implements SparkOrcValueWriter {
- private final int precision;
- private final int scale;
+ private static class Decimal38Writer implements OrcValueWriter<Decimal> {
- Decimal38Writer(int precision, int scale) {
- this.precision = precision;
- this.scale = scale;
+ @Override
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
+ ((DecimalColumnVector) output).vector[rowId].set(
+ HiveDecimal.create(decimal.toJavaBigDecimal()));
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((DecimalColumnVector) output).vector[rowId].set(
- HiveDecimal.create(data.getDecimal(column, precision, scale)
- .toJavaBigDecimal()));
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class ListWriter implements SparkOrcValueWriter {
- private final SparkOrcValueWriter writer;
+ private static class ListWriter implements OrcValueWriter<ArrayData> {
+ private final OrcValueWriter writer;
+ private final SparkOrcWriter.FieldGetter fieldGetter;
Review comment:
I see lots of warning in my intellij for this two lines, the following changes can address the warning:
```patch
diff --git a/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java b/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
index a56977ffd..41b8ccc35 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
@@ -57,11 +57,11 @@ class SparkOrcValueWriters {
}
static OrcValueWriter<?> list(OrcValueWriter<?> element, List<TypeDescription> orcType) {
- return new ListWriter(element, orcType);
+ return new ListWriter<>(element, orcType);
}
static OrcValueWriter<?> map(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter,
- List<TypeDescription> orcType) {
+ List<TypeDescription> orcType) {
return new MapWriter(keyWriter, valueWriter, orcType);
}
@@ -129,16 +129,17 @@ class SparkOrcValueWriters {
}
}
- private static class ListWriter implements OrcValueWriter<ArrayData> {
- private final OrcValueWriter writer;
- private final SparkOrcWriter.FieldGetter fieldGetter;
+ private static class ListWriter<T> implements OrcValueWriter<ArrayData> {
+ private final OrcValueWriter<T> writer;
+ private final SparkOrcWriter.FieldGetter<T> fieldGetter;
- ListWriter(OrcValueWriter<?> writer, List<TypeDescription> orcTypes) {
+ @SuppressWarnings("unchecked")
+ ListWriter(OrcValueWriter<T> writer, List<TypeDescription> orcTypes) {
if (orcTypes.size() != 1) {
throw new IllegalArgumentException("Expected one (and same) ORC type for list elements, got: " + orcTypes);
}
this.writer = writer;
- this.fieldGetter = SparkOrcWriter.createFieldGetter(orcTypes.get(0));
+ this.fieldGetter = (SparkOrcWriter.FieldGetter<T>) SparkOrcWriter.createFieldGetter(orcTypes.get(0));
}
@Override
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r725838169
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriter.java
##########
@@ -115,53 +116,72 @@ private WriteBuilder() {
}
@Override
- @SuppressWarnings("unchecked")
public void write(Record value, VectorizedRowBatch output) {
- Preconditions.checkArgument(writer instanceof RecordWriter, "writer must be a RecordWriter.");
+ Preconditions.checkArgument(value != null, "value must not be null");
int row = output.size;
output.size += 1;
- List<OrcValueWriter<?>> writers = ((RecordWriter) writer).writers();
- for (int c = 0; c < writers.size(); ++c) {
- OrcValueWriter child = writers.get(c);
- child.write(row, value.get(c, child.getJavaClass()), output.cols[c]);
- }
+
+ writer.rootNonNullWrite(row, value, output);
}
@Override
public Stream<FieldMetrics<?>> metrics() {
return writer.metrics();
}
- private static class RecordWriter implements OrcValueWriter<Record> {
+ public abstract static class StructWriter<S> implements OrcValueWriter<S> {
Review comment:
Copied the comment from here: https://github.com/apache/iceberg/pull/3250#discussion_r725771333
> This generic StructWriter doesn't belongs to the concrete OrcRowWriter<Record>, moving this to the writer definition class GenericOrcWriters looks more reasonable to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r726904122
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcWriter.java
##########
@@ -42,75 +46,72 @@
*/
public class SparkOrcWriter implements OrcRowWriter<InternalRow> {
- private final SparkOrcValueWriter writer;
+ private final InternalRowWriter writer;
public SparkOrcWriter(Schema iSchema, TypeDescription orcSchema) {
Preconditions.checkArgument(orcSchema.getCategory() == TypeDescription.Category.STRUCT,
"Top level must be a struct " + orcSchema);
- writer = OrcSchemaWithTypeVisitor.visit(iSchema, orcSchema, new WriteBuilder());
+ writer = (InternalRowWriter) OrcSchemaWithTypeVisitor.visit(iSchema, orcSchema, new WriteBuilder());
}
@Override
public void write(InternalRow value, VectorizedRowBatch output) {
- Preconditions.checkArgument(writer instanceof StructWriter, "writer must be StructWriter");
+ Preconditions.checkArgument(value != null, "value must not be null");
Review comment:
Okay, I just checked the flink/hive/spark write path, the `RowData`, `InternalRow` object cannot be allowed to set to be null, although all the field values from this row can set to be null. So this Preconditions#checkArgument looks good to me ! Thanks for you verification, @szlta !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r725845014
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriter.java
##########
@@ -115,53 +116,72 @@ private WriteBuilder() {
}
@Override
- @SuppressWarnings("unchecked")
public void write(Record value, VectorizedRowBatch output) {
- Preconditions.checkArgument(writer instanceof RecordWriter, "writer must be a RecordWriter.");
+ Preconditions.checkArgument(value != null, "value must not be null");
int row = output.size;
output.size += 1;
- List<OrcValueWriter<?>> writers = ((RecordWriter) writer).writers();
- for (int c = 0; c < writers.size(); ++c) {
- OrcValueWriter child = writers.get(c);
- child.write(row, value.get(c, child.getJavaClass()), output.cols[c]);
- }
+
+ writer.rootNonNullWrite(row, value, output);
}
@Override
public Stream<FieldMetrics<?>> metrics() {
return writer.metrics();
}
- private static class RecordWriter implements OrcValueWriter<Record> {
+ public abstract static class StructWriter<S> implements OrcValueWriter<S> {
private final List<OrcValueWriter<?>> writers;
- RecordWriter(List<OrcValueWriter<?>> writers) {
+ protected StructWriter(List<OrcValueWriter<?>> writers) {
this.writers = writers;
}
- List<OrcValueWriter<?>> writers() {
- return writers;
+ public List<OrcValueWriter<?>> writers() {
+ return this.writers;
}
@Override
- public Class<Record> getJavaClass() {
- return Record.class;
+ public Stream<FieldMetrics<?>> metrics() {
+ return writers.stream().flatMap(OrcValueWriter::metrics);
}
@Override
- @SuppressWarnings("unchecked")
- public void nonNullWrite(int rowId, Record data, ColumnVector output) {
+ public void nonNullWrite(int rowId, S value, ColumnVector output) {
StructColumnVector cv = (StructColumnVector) output;
+ write(rowId, value, c -> cv.fields[c]);
+ }
+
+ // Special case of writing the root struct
+ public void rootNonNullWrite(int rowId, S value, VectorizedRowBatch output) {
+ write(rowId, value, c -> output.cols[c]);
+ }
+
+ private void write(int rowId, S value, Function<Integer, ColumnVector> colVectorAtFunc) {
Review comment:
Nit: an `@SuppressWarnings("unchecked")` is required ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r727066952
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriters.java
##########
@@ -531,6 +534,46 @@ public void nonNullWrite(int rowId, Map<K, V> map, ColumnVector output) {
}
}
+ public abstract static class StructWriter<S> implements OrcValueWriter<S> {
+ private final List<OrcValueWriter<?>> writers;
+
+ protected StructWriter(List<OrcValueWriter<?>> writers) {
+ this.writers = writers;
+ }
+
+ public List<OrcValueWriter<?>> writers() {
Review comment:
OK
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szlta commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
szlta commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r727875963
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -19,245 +19,130 @@
package org.apache.iceberg.spark.data;
+import java.util.List;
import java.util.stream.Stream;
-import org.apache.iceberg.DoubleFieldMetrics;
import org.apache.iceberg.FieldMetrics;
-import org.apache.iceberg.FloatFieldMetrics;
+import org.apache.iceberg.orc.OrcValueWriter;
+import org.apache.orc.TypeDescription;
import org.apache.orc.storage.common.type.HiveDecimal;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
-import org.apache.orc.storage.ql.exec.vector.DoubleColumnVector;
import org.apache.orc.storage.ql.exec.vector.ListColumnVector;
-import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
import org.apache.orc.storage.ql.exec.vector.MapColumnVector;
import org.apache.orc.storage.ql.exec.vector.TimestampColumnVector;
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
import org.apache.spark.sql.catalyst.util.ArrayData;
import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
class SparkOrcValueWriters {
private SparkOrcValueWriters() {
}
- static SparkOrcValueWriter booleans() {
- return BooleanWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter bytes() {
- return ByteWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter shorts() {
- return ShortWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter ints() {
- return IntWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter longs() {
- return LongWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter floats(int id) {
- return new FloatWriter(id);
- }
-
- static SparkOrcValueWriter doubles(int id) {
- return new DoubleWriter(id);
- }
-
- static SparkOrcValueWriter byteArrays() {
- return BytesWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter strings() {
+ static OrcValueWriter<?> strings() {
return StringWriter.INSTANCE;
}
- static SparkOrcValueWriter timestampTz() {
+ static OrcValueWriter<?> timestampTz() {
return TimestampTzWriter.INSTANCE;
}
- static SparkOrcValueWriter decimal(int precision, int scale) {
+ static OrcValueWriter<?> decimal(int precision, int scale) {
if (precision <= 18) {
- return new Decimal18Writer(precision, scale);
+ return new Decimal18Writer(scale);
} else {
- return new Decimal38Writer(precision, scale);
- }
- }
-
- static SparkOrcValueWriter list(SparkOrcValueWriter element) {
- return new ListWriter(element);
- }
-
- static SparkOrcValueWriter map(SparkOrcValueWriter keyWriter, SparkOrcValueWriter valueWriter) {
- return new MapWriter(keyWriter, valueWriter);
- }
-
- private static class BooleanWriter implements SparkOrcValueWriter {
- private static final BooleanWriter INSTANCE = new BooleanWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getBoolean(column) ? 1 : 0;
- }
- }
-
- private static class ByteWriter implements SparkOrcValueWriter {
- private static final ByteWriter INSTANCE = new ByteWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getByte(column);
- }
- }
-
- private static class ShortWriter implements SparkOrcValueWriter {
- private static final ShortWriter INSTANCE = new ShortWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getShort(column);
+ return new Decimal38Writer();
}
}
- private static class IntWriter implements SparkOrcValueWriter {
- private static final IntWriter INSTANCE = new IntWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getInt(column);
- }
+ static OrcValueWriter<?> list(OrcValueWriter<?> element, List<TypeDescription> orcType) {
+ return new ListWriter(element, orcType);
}
- private static class LongWriter implements SparkOrcValueWriter {
- private static final LongWriter INSTANCE = new LongWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getLong(column);
- }
+ static OrcValueWriter<?> map(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter,
+ List<TypeDescription> orcType) {
+ return new MapWriter(keyWriter, valueWriter, orcType);
}
- private static class FloatWriter implements SparkOrcValueWriter {
- private final FloatFieldMetrics.Builder floatFieldMetricsBuilder;
-
- private FloatWriter(int id) {
- this.floatFieldMetricsBuilder = new FloatFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- float floatValue = data.getFloat(column);
- ((DoubleColumnVector) output).vector[rowId] = floatValue;
- floatFieldMetricsBuilder.addValue(floatValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(floatFieldMetricsBuilder.build());
- }
- }
-
- private static class DoubleWriter implements SparkOrcValueWriter {
- private final DoubleFieldMetrics.Builder doubleFieldMetricsBuilder;
-
- private DoubleWriter(int id) {
- this.doubleFieldMetricsBuilder = new DoubleFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- double doubleValue = data.getDouble(column);
- ((DoubleColumnVector) output).vector[rowId] = doubleValue;
- doubleFieldMetricsBuilder.addValue(doubleValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(doubleFieldMetricsBuilder.build());
- }
- }
-
- private static class BytesWriter implements SparkOrcValueWriter {
- private static final BytesWriter INSTANCE = new BytesWriter();
+ private static class StringWriter implements OrcValueWriter<UTF8String> {
+ private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- // getBinary always makes a copy, so we don't need to worry about it
- // being changed behind our back.
- byte[] value = data.getBinary(column);
+ public void nonNullWrite(int rowId, UTF8String data, ColumnVector output) {
+ byte[] value = data.getBytes();
((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
}
- }
-
- private static class StringWriter implements SparkOrcValueWriter {
- private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- byte[] value = data.getUTF8String(column).getBytes();
- ((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
+ public Class<?> getJavaClass() {
+ return UTF8String.class;
}
}
- private static class TimestampTzWriter implements SparkOrcValueWriter {
+ private static class TimestampTzWriter implements OrcValueWriter<Long> {
private static final TimestampTzWriter INSTANCE = new TimestampTzWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Long micros, ColumnVector output) {
TimestampColumnVector cv = (TimestampColumnVector) output;
- long micros = data.getLong(column); // it could be negative.
cv.time[rowId] = Math.floorDiv(micros, 1_000); // millis
cv.nanos[rowId] = (int) Math.floorMod(micros, 1_000_000) * 1_000; // nanos
}
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Long.class;
+ }
}
- private static class Decimal18Writer implements SparkOrcValueWriter {
- private final int precision;
+ private static class Decimal18Writer implements OrcValueWriter<Decimal> {
private final int scale;
- Decimal18Writer(int precision, int scale) {
- this.precision = precision;
+ Decimal18Writer(int scale) {
this.scale = scale;
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
((DecimalColumnVector) output).vector[rowId].setFromLongAndScale(
- data.getDecimal(column, precision, scale).toUnscaledLong(), scale);
+ decimal.toUnscaledLong(), scale);
+ }
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class Decimal38Writer implements SparkOrcValueWriter {
- private final int precision;
- private final int scale;
+ private static class Decimal38Writer implements OrcValueWriter<Decimal> {
- Decimal38Writer(int precision, int scale) {
- this.precision = precision;
- this.scale = scale;
+ @Override
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
+ ((DecimalColumnVector) output).vector[rowId].set(
+ HiveDecimal.create(decimal.toJavaBigDecimal()));
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((DecimalColumnVector) output).vector[rowId].set(
- HiveDecimal.create(data.getDecimal(column, precision, scale)
- .toJavaBigDecimal()));
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class ListWriter implements SparkOrcValueWriter {
- private final SparkOrcValueWriter writer;
+ private static class ListWriter implements OrcValueWriter<ArrayData> {
+ private final OrcValueWriter writer;
+ private final SparkOrcWriter.FieldGetter fieldGetter;
- ListWriter(SparkOrcValueWriter writer) {
+ ListWriter(OrcValueWriter<?> writer, List<TypeDescription> orcTypes) {
Review comment:
We could do that, but I wanted to keep the symmetry in the OrcSchemWithTypeVisitor side:
https://github.com/apache/iceberg/pull/3248/files#diff-d3255f1d9dad8c538687fe8a94d585736b565cccb0952a9a5595d1ef9b952b93R73-R87
So for any complex ORC type (i.e. struct, list, map) I'm supplying the inner ORC type by calling .getChildren() for all these cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx merged pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
openinx merged pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r726971730
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriters.java
##########
@@ -531,6 +534,46 @@ public void nonNullWrite(int rowId, Map<K, V> map, ColumnVector output) {
}
}
+ public abstract static class StructWriter<S> implements OrcValueWriter<S> {
+ private final List<OrcValueWriter<?>> writers;
+
+ protected StructWriter(List<OrcValueWriter<?>> writers) {
+ this.writers = writers;
+ }
+
+ public List<OrcValueWriter<?>> writers() {
Review comment:
Nit: Maybe we could remove this method now because the `InternalRowWriter` don't need this any more if we plan to address this comment: https://github.com/apache/iceberg/pull/3248/files#diff-d3255f1d9dad8c538687fe8a94d585736b565cccb0952a9a5595d1ef9b952b93R144
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r727053434
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -19,245 +19,130 @@
package org.apache.iceberg.spark.data;
+import java.util.List;
import java.util.stream.Stream;
-import org.apache.iceberg.DoubleFieldMetrics;
import org.apache.iceberg.FieldMetrics;
-import org.apache.iceberg.FloatFieldMetrics;
+import org.apache.iceberg.orc.OrcValueWriter;
+import org.apache.orc.TypeDescription;
import org.apache.orc.storage.common.type.HiveDecimal;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
-import org.apache.orc.storage.ql.exec.vector.DoubleColumnVector;
import org.apache.orc.storage.ql.exec.vector.ListColumnVector;
-import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
import org.apache.orc.storage.ql.exec.vector.MapColumnVector;
import org.apache.orc.storage.ql.exec.vector.TimestampColumnVector;
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
import org.apache.spark.sql.catalyst.util.ArrayData;
import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
class SparkOrcValueWriters {
private SparkOrcValueWriters() {
}
- static SparkOrcValueWriter booleans() {
- return BooleanWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter bytes() {
- return ByteWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter shorts() {
- return ShortWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter ints() {
- return IntWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter longs() {
- return LongWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter floats(int id) {
- return new FloatWriter(id);
- }
-
- static SparkOrcValueWriter doubles(int id) {
- return new DoubleWriter(id);
- }
-
- static SparkOrcValueWriter byteArrays() {
- return BytesWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter strings() {
+ static OrcValueWriter<?> strings() {
return StringWriter.INSTANCE;
}
- static SparkOrcValueWriter timestampTz() {
+ static OrcValueWriter<?> timestampTz() {
return TimestampTzWriter.INSTANCE;
}
- static SparkOrcValueWriter decimal(int precision, int scale) {
+ static OrcValueWriter<?> decimal(int precision, int scale) {
if (precision <= 18) {
- return new Decimal18Writer(precision, scale);
+ return new Decimal18Writer(scale);
} else {
- return new Decimal38Writer(precision, scale);
- }
- }
-
- static SparkOrcValueWriter list(SparkOrcValueWriter element) {
- return new ListWriter(element);
- }
-
- static SparkOrcValueWriter map(SparkOrcValueWriter keyWriter, SparkOrcValueWriter valueWriter) {
- return new MapWriter(keyWriter, valueWriter);
- }
-
- private static class BooleanWriter implements SparkOrcValueWriter {
- private static final BooleanWriter INSTANCE = new BooleanWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getBoolean(column) ? 1 : 0;
- }
- }
-
- private static class ByteWriter implements SparkOrcValueWriter {
- private static final ByteWriter INSTANCE = new ByteWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getByte(column);
- }
- }
-
- private static class ShortWriter implements SparkOrcValueWriter {
- private static final ShortWriter INSTANCE = new ShortWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getShort(column);
+ return new Decimal38Writer();
}
}
- private static class IntWriter implements SparkOrcValueWriter {
- private static final IntWriter INSTANCE = new IntWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getInt(column);
- }
+ static OrcValueWriter<?> list(OrcValueWriter<?> element, List<TypeDescription> orcType) {
+ return new ListWriter(element, orcType);
}
- private static class LongWriter implements SparkOrcValueWriter {
- private static final LongWriter INSTANCE = new LongWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getLong(column);
- }
+ static OrcValueWriter<?> map(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter,
+ List<TypeDescription> orcType) {
+ return new MapWriter(keyWriter, valueWriter, orcType);
}
- private static class FloatWriter implements SparkOrcValueWriter {
- private final FloatFieldMetrics.Builder floatFieldMetricsBuilder;
-
- private FloatWriter(int id) {
- this.floatFieldMetricsBuilder = new FloatFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- float floatValue = data.getFloat(column);
- ((DoubleColumnVector) output).vector[rowId] = floatValue;
- floatFieldMetricsBuilder.addValue(floatValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(floatFieldMetricsBuilder.build());
- }
- }
-
- private static class DoubleWriter implements SparkOrcValueWriter {
- private final DoubleFieldMetrics.Builder doubleFieldMetricsBuilder;
-
- private DoubleWriter(int id) {
- this.doubleFieldMetricsBuilder = new DoubleFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- double doubleValue = data.getDouble(column);
- ((DoubleColumnVector) output).vector[rowId] = doubleValue;
- doubleFieldMetricsBuilder.addValue(doubleValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(doubleFieldMetricsBuilder.build());
- }
- }
-
- private static class BytesWriter implements SparkOrcValueWriter {
- private static final BytesWriter INSTANCE = new BytesWriter();
+ private static class StringWriter implements OrcValueWriter<UTF8String> {
+ private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- // getBinary always makes a copy, so we don't need to worry about it
- // being changed behind our back.
- byte[] value = data.getBinary(column);
+ public void nonNullWrite(int rowId, UTF8String data, ColumnVector output) {
+ byte[] value = data.getBytes();
((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
}
- }
-
- private static class StringWriter implements SparkOrcValueWriter {
- private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- byte[] value = data.getUTF8String(column).getBytes();
- ((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
+ public Class<?> getJavaClass() {
+ return UTF8String.class;
}
}
- private static class TimestampTzWriter implements SparkOrcValueWriter {
+ private static class TimestampTzWriter implements OrcValueWriter<Long> {
private static final TimestampTzWriter INSTANCE = new TimestampTzWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Long micros, ColumnVector output) {
TimestampColumnVector cv = (TimestampColumnVector) output;
- long micros = data.getLong(column); // it could be negative.
cv.time[rowId] = Math.floorDiv(micros, 1_000); // millis
cv.nanos[rowId] = (int) Math.floorMod(micros, 1_000_000) * 1_000; // nanos
}
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Long.class;
+ }
}
- private static class Decimal18Writer implements SparkOrcValueWriter {
- private final int precision;
+ private static class Decimal18Writer implements OrcValueWriter<Decimal> {
private final int scale;
- Decimal18Writer(int precision, int scale) {
- this.precision = precision;
+ Decimal18Writer(int scale) {
this.scale = scale;
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
((DecimalColumnVector) output).vector[rowId].setFromLongAndScale(
- data.getDecimal(column, precision, scale).toUnscaledLong(), scale);
+ decimal.toUnscaledLong(), scale);
+ }
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class Decimal38Writer implements SparkOrcValueWriter {
- private final int precision;
- private final int scale;
+ private static class Decimal38Writer implements OrcValueWriter<Decimal> {
- Decimal38Writer(int precision, int scale) {
- this.precision = precision;
- this.scale = scale;
+ @Override
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
+ ((DecimalColumnVector) output).vector[rowId].set(
+ HiveDecimal.create(decimal.toJavaBigDecimal()));
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((DecimalColumnVector) output).vector[rowId].set(
- HiveDecimal.create(data.getDecimal(column, precision, scale)
- .toJavaBigDecimal()));
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class ListWriter implements SparkOrcValueWriter {
- private final SparkOrcValueWriter writer;
+ private static class ListWriter implements OrcValueWriter<ArrayData> {
+ private final OrcValueWriter writer;
+ private final SparkOrcWriter.FieldGetter fieldGetter;
- ListWriter(SparkOrcValueWriter writer) {
+ ListWriter(OrcValueWriter<?> writer, List<TypeDescription> orcTypes) {
Review comment:
How about using the `elemType` as the TypeDescription argument because that looks more symmetrical to the writer? I think we can move orcTypes size checking out of this scope.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szlta commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
szlta commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r727872599
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -19,245 +19,130 @@
package org.apache.iceberg.spark.data;
+import java.util.List;
import java.util.stream.Stream;
-import org.apache.iceberg.DoubleFieldMetrics;
import org.apache.iceberg.FieldMetrics;
-import org.apache.iceberg.FloatFieldMetrics;
+import org.apache.iceberg.orc.OrcValueWriter;
+import org.apache.orc.TypeDescription;
import org.apache.orc.storage.common.type.HiveDecimal;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
-import org.apache.orc.storage.ql.exec.vector.DoubleColumnVector;
import org.apache.orc.storage.ql.exec.vector.ListColumnVector;
-import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
import org.apache.orc.storage.ql.exec.vector.MapColumnVector;
import org.apache.orc.storage.ql.exec.vector.TimestampColumnVector;
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
import org.apache.spark.sql.catalyst.util.ArrayData;
import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
class SparkOrcValueWriters {
private SparkOrcValueWriters() {
}
- static SparkOrcValueWriter booleans() {
- return BooleanWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter bytes() {
- return ByteWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter shorts() {
- return ShortWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter ints() {
- return IntWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter longs() {
- return LongWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter floats(int id) {
- return new FloatWriter(id);
- }
-
- static SparkOrcValueWriter doubles(int id) {
- return new DoubleWriter(id);
- }
-
- static SparkOrcValueWriter byteArrays() {
- return BytesWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter strings() {
+ static OrcValueWriter<?> strings() {
return StringWriter.INSTANCE;
}
- static SparkOrcValueWriter timestampTz() {
+ static OrcValueWriter<?> timestampTz() {
return TimestampTzWriter.INSTANCE;
}
- static SparkOrcValueWriter decimal(int precision, int scale) {
+ static OrcValueWriter<?> decimal(int precision, int scale) {
if (precision <= 18) {
- return new Decimal18Writer(precision, scale);
+ return new Decimal18Writer(scale);
} else {
- return new Decimal38Writer(precision, scale);
- }
- }
-
- static SparkOrcValueWriter list(SparkOrcValueWriter element) {
- return new ListWriter(element);
- }
-
- static SparkOrcValueWriter map(SparkOrcValueWriter keyWriter, SparkOrcValueWriter valueWriter) {
- return new MapWriter(keyWriter, valueWriter);
- }
-
- private static class BooleanWriter implements SparkOrcValueWriter {
- private static final BooleanWriter INSTANCE = new BooleanWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getBoolean(column) ? 1 : 0;
- }
- }
-
- private static class ByteWriter implements SparkOrcValueWriter {
- private static final ByteWriter INSTANCE = new ByteWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getByte(column);
- }
- }
-
- private static class ShortWriter implements SparkOrcValueWriter {
- private static final ShortWriter INSTANCE = new ShortWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getShort(column);
+ return new Decimal38Writer();
}
}
- private static class IntWriter implements SparkOrcValueWriter {
- private static final IntWriter INSTANCE = new IntWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getInt(column);
- }
+ static OrcValueWriter<?> list(OrcValueWriter<?> element, List<TypeDescription> orcType) {
+ return new ListWriter(element, orcType);
}
- private static class LongWriter implements SparkOrcValueWriter {
- private static final LongWriter INSTANCE = new LongWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getLong(column);
- }
+ static OrcValueWriter<?> map(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter,
+ List<TypeDescription> orcType) {
Review comment:
It actually should be named orcTypes - I'll correct that in the next round.
It is required so that we can determine the two fieldGetter types that will be used for retrieving the keys and values of the MapData object. This works similarly to the ListWriter.
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -19,245 +19,130 @@
package org.apache.iceberg.spark.data;
+import java.util.List;
import java.util.stream.Stream;
-import org.apache.iceberg.DoubleFieldMetrics;
import org.apache.iceberg.FieldMetrics;
-import org.apache.iceberg.FloatFieldMetrics;
+import org.apache.iceberg.orc.OrcValueWriter;
+import org.apache.orc.TypeDescription;
import org.apache.orc.storage.common.type.HiveDecimal;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
-import org.apache.orc.storage.ql.exec.vector.DoubleColumnVector;
import org.apache.orc.storage.ql.exec.vector.ListColumnVector;
-import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
import org.apache.orc.storage.ql.exec.vector.MapColumnVector;
import org.apache.orc.storage.ql.exec.vector.TimestampColumnVector;
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
import org.apache.spark.sql.catalyst.util.ArrayData;
import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
class SparkOrcValueWriters {
private SparkOrcValueWriters() {
}
- static SparkOrcValueWriter booleans() {
- return BooleanWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter bytes() {
- return ByteWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter shorts() {
- return ShortWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter ints() {
- return IntWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter longs() {
- return LongWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter floats(int id) {
- return new FloatWriter(id);
- }
-
- static SparkOrcValueWriter doubles(int id) {
- return new DoubleWriter(id);
- }
-
- static SparkOrcValueWriter byteArrays() {
- return BytesWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter strings() {
+ static OrcValueWriter<?> strings() {
return StringWriter.INSTANCE;
}
- static SparkOrcValueWriter timestampTz() {
+ static OrcValueWriter<?> timestampTz() {
return TimestampTzWriter.INSTANCE;
}
- static SparkOrcValueWriter decimal(int precision, int scale) {
+ static OrcValueWriter<?> decimal(int precision, int scale) {
if (precision <= 18) {
- return new Decimal18Writer(precision, scale);
+ return new Decimal18Writer(scale);
} else {
- return new Decimal38Writer(precision, scale);
- }
- }
-
- static SparkOrcValueWriter list(SparkOrcValueWriter element) {
- return new ListWriter(element);
- }
-
- static SparkOrcValueWriter map(SparkOrcValueWriter keyWriter, SparkOrcValueWriter valueWriter) {
- return new MapWriter(keyWriter, valueWriter);
- }
-
- private static class BooleanWriter implements SparkOrcValueWriter {
- private static final BooleanWriter INSTANCE = new BooleanWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getBoolean(column) ? 1 : 0;
- }
- }
-
- private static class ByteWriter implements SparkOrcValueWriter {
- private static final ByteWriter INSTANCE = new ByteWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getByte(column);
- }
- }
-
- private static class ShortWriter implements SparkOrcValueWriter {
- private static final ShortWriter INSTANCE = new ShortWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getShort(column);
+ return new Decimal38Writer();
}
}
- private static class IntWriter implements SparkOrcValueWriter {
- private static final IntWriter INSTANCE = new IntWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getInt(column);
- }
+ static OrcValueWriter<?> list(OrcValueWriter<?> element, List<TypeDescription> orcType) {
+ return new ListWriter(element, orcType);
}
- private static class LongWriter implements SparkOrcValueWriter {
- private static final LongWriter INSTANCE = new LongWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getLong(column);
- }
+ static OrcValueWriter<?> map(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter,
+ List<TypeDescription> orcType) {
+ return new MapWriter(keyWriter, valueWriter, orcType);
}
- private static class FloatWriter implements SparkOrcValueWriter {
- private final FloatFieldMetrics.Builder floatFieldMetricsBuilder;
-
- private FloatWriter(int id) {
- this.floatFieldMetricsBuilder = new FloatFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- float floatValue = data.getFloat(column);
- ((DoubleColumnVector) output).vector[rowId] = floatValue;
- floatFieldMetricsBuilder.addValue(floatValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(floatFieldMetricsBuilder.build());
- }
- }
-
- private static class DoubleWriter implements SparkOrcValueWriter {
- private final DoubleFieldMetrics.Builder doubleFieldMetricsBuilder;
-
- private DoubleWriter(int id) {
- this.doubleFieldMetricsBuilder = new DoubleFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- double doubleValue = data.getDouble(column);
- ((DoubleColumnVector) output).vector[rowId] = doubleValue;
- doubleFieldMetricsBuilder.addValue(doubleValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(doubleFieldMetricsBuilder.build());
- }
- }
-
- private static class BytesWriter implements SparkOrcValueWriter {
- private static final BytesWriter INSTANCE = new BytesWriter();
+ private static class StringWriter implements OrcValueWriter<UTF8String> {
+ private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- // getBinary always makes a copy, so we don't need to worry about it
- // being changed behind our back.
- byte[] value = data.getBinary(column);
+ public void nonNullWrite(int rowId, UTF8String data, ColumnVector output) {
+ byte[] value = data.getBytes();
((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
}
- }
-
- private static class StringWriter implements SparkOrcValueWriter {
- private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- byte[] value = data.getUTF8String(column).getBytes();
- ((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
+ public Class<?> getJavaClass() {
+ return UTF8String.class;
}
}
- private static class TimestampTzWriter implements SparkOrcValueWriter {
+ private static class TimestampTzWriter implements OrcValueWriter<Long> {
private static final TimestampTzWriter INSTANCE = new TimestampTzWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Long micros, ColumnVector output) {
TimestampColumnVector cv = (TimestampColumnVector) output;
- long micros = data.getLong(column); // it could be negative.
cv.time[rowId] = Math.floorDiv(micros, 1_000); // millis
cv.nanos[rowId] = (int) Math.floorMod(micros, 1_000_000) * 1_000; // nanos
}
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Long.class;
+ }
}
- private static class Decimal18Writer implements SparkOrcValueWriter {
- private final int precision;
+ private static class Decimal18Writer implements OrcValueWriter<Decimal> {
private final int scale;
- Decimal18Writer(int precision, int scale) {
- this.precision = precision;
+ Decimal18Writer(int scale) {
this.scale = scale;
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
((DecimalColumnVector) output).vector[rowId].setFromLongAndScale(
- data.getDecimal(column, precision, scale).toUnscaledLong(), scale);
+ decimal.toUnscaledLong(), scale);
+ }
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class Decimal38Writer implements SparkOrcValueWriter {
- private final int precision;
- private final int scale;
+ private static class Decimal38Writer implements OrcValueWriter<Decimal> {
- Decimal38Writer(int precision, int scale) {
- this.precision = precision;
- this.scale = scale;
+ @Override
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
+ ((DecimalColumnVector) output).vector[rowId].set(
+ HiveDecimal.create(decimal.toJavaBigDecimal()));
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((DecimalColumnVector) output).vector[rowId].set(
- HiveDecimal.create(data.getDecimal(column, precision, scale)
- .toJavaBigDecimal()));
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class ListWriter implements SparkOrcValueWriter {
- private final SparkOrcValueWriter writer;
+ private static class ListWriter implements OrcValueWriter<ArrayData> {
+ private final OrcValueWriter writer;
+ private final SparkOrcWriter.FieldGetter fieldGetter;
- ListWriter(SparkOrcValueWriter writer) {
+ ListWriter(OrcValueWriter<?> writer, List<TypeDescription> orcTypes) {
Review comment:
We could do that, but I wanted to keep the symmetry in the OrcSchemWithTypeVisitor side:
https://github.com/apache/iceberg/pull/3248/files#diff-d3255f1d9dad8c538687fe8a94d585736b565cccb0952a9a5595d1ef9b952b93R73-R87
So for any complex ORC type (i.e. struct, list, map) I'm supplying the inner ORC type by calling .getChildren() for all these cases.
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -19,245 +19,130 @@
package org.apache.iceberg.spark.data;
+import java.util.List;
import java.util.stream.Stream;
-import org.apache.iceberg.DoubleFieldMetrics;
import org.apache.iceberg.FieldMetrics;
-import org.apache.iceberg.FloatFieldMetrics;
+import org.apache.iceberg.orc.OrcValueWriter;
+import org.apache.orc.TypeDescription;
import org.apache.orc.storage.common.type.HiveDecimal;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
-import org.apache.orc.storage.ql.exec.vector.DoubleColumnVector;
import org.apache.orc.storage.ql.exec.vector.ListColumnVector;
-import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
import org.apache.orc.storage.ql.exec.vector.MapColumnVector;
import org.apache.orc.storage.ql.exec.vector.TimestampColumnVector;
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
import org.apache.spark.sql.catalyst.util.ArrayData;
import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
class SparkOrcValueWriters {
private SparkOrcValueWriters() {
}
- static SparkOrcValueWriter booleans() {
- return BooleanWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter bytes() {
- return ByteWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter shorts() {
- return ShortWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter ints() {
- return IntWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter longs() {
- return LongWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter floats(int id) {
- return new FloatWriter(id);
- }
-
- static SparkOrcValueWriter doubles(int id) {
- return new DoubleWriter(id);
- }
-
- static SparkOrcValueWriter byteArrays() {
- return BytesWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter strings() {
+ static OrcValueWriter<?> strings() {
return StringWriter.INSTANCE;
}
- static SparkOrcValueWriter timestampTz() {
+ static OrcValueWriter<?> timestampTz() {
return TimestampTzWriter.INSTANCE;
}
- static SparkOrcValueWriter decimal(int precision, int scale) {
+ static OrcValueWriter<?> decimal(int precision, int scale) {
if (precision <= 18) {
- return new Decimal18Writer(precision, scale);
+ return new Decimal18Writer(scale);
} else {
- return new Decimal38Writer(precision, scale);
- }
- }
-
- static SparkOrcValueWriter list(SparkOrcValueWriter element) {
- return new ListWriter(element);
- }
-
- static SparkOrcValueWriter map(SparkOrcValueWriter keyWriter, SparkOrcValueWriter valueWriter) {
- return new MapWriter(keyWriter, valueWriter);
- }
-
- private static class BooleanWriter implements SparkOrcValueWriter {
- private static final BooleanWriter INSTANCE = new BooleanWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getBoolean(column) ? 1 : 0;
- }
- }
-
- private static class ByteWriter implements SparkOrcValueWriter {
- private static final ByteWriter INSTANCE = new ByteWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getByte(column);
- }
- }
-
- private static class ShortWriter implements SparkOrcValueWriter {
- private static final ShortWriter INSTANCE = new ShortWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getShort(column);
+ return new Decimal38Writer();
}
}
- private static class IntWriter implements SparkOrcValueWriter {
- private static final IntWriter INSTANCE = new IntWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getInt(column);
- }
+ static OrcValueWriter<?> list(OrcValueWriter<?> element, List<TypeDescription> orcType) {
+ return new ListWriter(element, orcType);
}
- private static class LongWriter implements SparkOrcValueWriter {
- private static final LongWriter INSTANCE = new LongWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getLong(column);
- }
+ static OrcValueWriter<?> map(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter,
+ List<TypeDescription> orcType) {
+ return new MapWriter(keyWriter, valueWriter, orcType);
}
- private static class FloatWriter implements SparkOrcValueWriter {
- private final FloatFieldMetrics.Builder floatFieldMetricsBuilder;
-
- private FloatWriter(int id) {
- this.floatFieldMetricsBuilder = new FloatFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- float floatValue = data.getFloat(column);
- ((DoubleColumnVector) output).vector[rowId] = floatValue;
- floatFieldMetricsBuilder.addValue(floatValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(floatFieldMetricsBuilder.build());
- }
- }
-
- private static class DoubleWriter implements SparkOrcValueWriter {
- private final DoubleFieldMetrics.Builder doubleFieldMetricsBuilder;
-
- private DoubleWriter(int id) {
- this.doubleFieldMetricsBuilder = new DoubleFieldMetrics.Builder(id);
- }
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- double doubleValue = data.getDouble(column);
- ((DoubleColumnVector) output).vector[rowId] = doubleValue;
- doubleFieldMetricsBuilder.addValue(doubleValue);
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return Stream.of(doubleFieldMetricsBuilder.build());
- }
- }
-
- private static class BytesWriter implements SparkOrcValueWriter {
- private static final BytesWriter INSTANCE = new BytesWriter();
+ private static class StringWriter implements OrcValueWriter<UTF8String> {
+ private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- // getBinary always makes a copy, so we don't need to worry about it
- // being changed behind our back.
- byte[] value = data.getBinary(column);
+ public void nonNullWrite(int rowId, UTF8String data, ColumnVector output) {
+ byte[] value = data.getBytes();
((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
}
- }
-
- private static class StringWriter implements SparkOrcValueWriter {
- private static final StringWriter INSTANCE = new StringWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- byte[] value = data.getUTF8String(column).getBytes();
- ((BytesColumnVector) output).setRef(rowId, value, 0, value.length);
+ public Class<?> getJavaClass() {
+ return UTF8String.class;
}
}
- private static class TimestampTzWriter implements SparkOrcValueWriter {
+ private static class TimestampTzWriter implements OrcValueWriter<Long> {
private static final TimestampTzWriter INSTANCE = new TimestampTzWriter();
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Long micros, ColumnVector output) {
TimestampColumnVector cv = (TimestampColumnVector) output;
- long micros = data.getLong(column); // it could be negative.
cv.time[rowId] = Math.floorDiv(micros, 1_000); // millis
cv.nanos[rowId] = (int) Math.floorMod(micros, 1_000_000) * 1_000; // nanos
}
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Long.class;
+ }
}
- private static class Decimal18Writer implements SparkOrcValueWriter {
- private final int precision;
+ private static class Decimal18Writer implements OrcValueWriter<Decimal> {
private final int scale;
- Decimal18Writer(int precision, int scale) {
- this.precision = precision;
+ Decimal18Writer(int scale) {
this.scale = scale;
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
((DecimalColumnVector) output).vector[rowId].setFromLongAndScale(
- data.getDecimal(column, precision, scale).toUnscaledLong(), scale);
+ decimal.toUnscaledLong(), scale);
+ }
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class Decimal38Writer implements SparkOrcValueWriter {
- private final int precision;
- private final int scale;
+ private static class Decimal38Writer implements OrcValueWriter<Decimal> {
- Decimal38Writer(int precision, int scale) {
- this.precision = precision;
- this.scale = scale;
+ @Override
+ public void nonNullWrite(int rowId, Decimal decimal, ColumnVector output) {
+ ((DecimalColumnVector) output).vector[rowId].set(
+ HiveDecimal.create(decimal.toJavaBigDecimal()));
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((DecimalColumnVector) output).vector[rowId].set(
- HiveDecimal.create(data.getDecimal(column, precision, scale)
- .toJavaBigDecimal()));
+ public Class<?> getJavaClass() {
+ return Decimal.class;
}
}
- private static class ListWriter implements SparkOrcValueWriter {
- private final SparkOrcValueWriter writer;
+ private static class ListWriter implements OrcValueWriter<ArrayData> {
+ private final OrcValueWriter writer;
+ private final SparkOrcWriter.FieldGetter fieldGetter;
Review comment:
Sounds good! Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szlta commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
szlta commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r726040970
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcWriter.java
##########
@@ -42,75 +46,72 @@
*/
public class SparkOrcWriter implements OrcRowWriter<InternalRow> {
- private final SparkOrcValueWriter writer;
+ private final InternalRowWriter writer;
public SparkOrcWriter(Schema iSchema, TypeDescription orcSchema) {
Preconditions.checkArgument(orcSchema.getCategory() == TypeDescription.Category.STRUCT,
"Top level must be a struct " + orcSchema);
- writer = OrcSchemaWithTypeVisitor.visit(iSchema, orcSchema, new WriteBuilder());
+ writer = (InternalRowWriter) OrcSchemaWithTypeVisitor.visit(iSchema, orcSchema, new WriteBuilder());
}
@Override
public void write(InternalRow value, VectorizedRowBatch output) {
- Preconditions.checkArgument(writer instanceof StructWriter, "writer must be StructWriter");
+ Preconditions.checkArgument(value != null, "value must not be null");
Review comment:
The variable "value" cannot be null in the original version either. If it was null, it would get passed here: https://github.com/apache/iceberg/blob/master/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcWriter.java#L63
..and would throw an NPE here: https://github.com/apache/iceberg/blob/master/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriter.java#L38
After my change we just throw a nicer exception for the same situation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r726912272
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriter.java
##########
@@ -150,18 +135,8 @@ public void write(Record value, VectorizedRowBatch output) {
}
@Override
- @SuppressWarnings("unchecked")
- public void nonNullWrite(int rowId, Record data, ColumnVector output) {
- StructColumnVector cv = (StructColumnVector) output;
- for (int c = 0; c < writers.size(); ++c) {
- OrcValueWriter child = writers.get(c);
- child.write(rowId, data.get(c, child.getJavaClass()), cv.fields[c]);
- }
- }
-
- @Override
- public Stream<FieldMetrics<?>> metrics() {
- return writers.stream().flatMap(OrcValueWriter::metrics);
+ protected Object get(Record struct, int index) {
+ return struct.get(index, writers().get(index).getJavaClass());
Review comment:
I think the `Class<?> getJavaClass()` from `OrcValueWriter` can be removed now ha ? Because this line is the only place that we will use the `getJavaClass` , while the `struct` is actually a `Record` which means we could just use the `struct#get(index)` here, right ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#issuecomment-940944552
Thanks @szlta for the contribution ! The PR almost looks good to me overall, just left several comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r726980379
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -19,245 +19,130 @@
package org.apache.iceberg.spark.data;
+import java.util.List;
import java.util.stream.Stream;
-import org.apache.iceberg.DoubleFieldMetrics;
import org.apache.iceberg.FieldMetrics;
-import org.apache.iceberg.FloatFieldMetrics;
+import org.apache.iceberg.orc.OrcValueWriter;
+import org.apache.orc.TypeDescription;
import org.apache.orc.storage.common.type.HiveDecimal;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
-import org.apache.orc.storage.ql.exec.vector.DoubleColumnVector;
import org.apache.orc.storage.ql.exec.vector.ListColumnVector;
-import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
import org.apache.orc.storage.ql.exec.vector.MapColumnVector;
import org.apache.orc.storage.ql.exec.vector.TimestampColumnVector;
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
import org.apache.spark.sql.catalyst.util.ArrayData;
import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
class SparkOrcValueWriters {
private SparkOrcValueWriters() {
}
- static SparkOrcValueWriter booleans() {
- return BooleanWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter bytes() {
- return ByteWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter shorts() {
- return ShortWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter ints() {
- return IntWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter longs() {
- return LongWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter floats(int id) {
- return new FloatWriter(id);
- }
-
- static SparkOrcValueWriter doubles(int id) {
- return new DoubleWriter(id);
- }
-
- static SparkOrcValueWriter byteArrays() {
- return BytesWriter.INSTANCE;
- }
-
- static SparkOrcValueWriter strings() {
+ static OrcValueWriter<?> strings() {
return StringWriter.INSTANCE;
}
- static SparkOrcValueWriter timestampTz() {
+ static OrcValueWriter<?> timestampTz() {
return TimestampTzWriter.INSTANCE;
}
- static SparkOrcValueWriter decimal(int precision, int scale) {
+ static OrcValueWriter<?> decimal(int precision, int scale) {
if (precision <= 18) {
- return new Decimal18Writer(precision, scale);
+ return new Decimal18Writer(scale);
} else {
- return new Decimal38Writer(precision, scale);
- }
- }
-
- static SparkOrcValueWriter list(SparkOrcValueWriter element) {
- return new ListWriter(element);
- }
-
- static SparkOrcValueWriter map(SparkOrcValueWriter keyWriter, SparkOrcValueWriter valueWriter) {
- return new MapWriter(keyWriter, valueWriter);
- }
-
- private static class BooleanWriter implements SparkOrcValueWriter {
- private static final BooleanWriter INSTANCE = new BooleanWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getBoolean(column) ? 1 : 0;
- }
- }
-
- private static class ByteWriter implements SparkOrcValueWriter {
- private static final ByteWriter INSTANCE = new ByteWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getByte(column);
- }
- }
-
- private static class ShortWriter implements SparkOrcValueWriter {
- private static final ShortWriter INSTANCE = new ShortWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getShort(column);
+ return new Decimal38Writer();
}
}
- private static class IntWriter implements SparkOrcValueWriter {
- private static final IntWriter INSTANCE = new IntWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getInt(column);
- }
+ static OrcValueWriter<?> list(OrcValueWriter<?> element, List<TypeDescription> orcType) {
+ return new ListWriter(element, orcType);
}
- private static class LongWriter implements SparkOrcValueWriter {
- private static final LongWriter INSTANCE = new LongWriter();
-
- @Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- ((LongColumnVector) output).vector[rowId] = data.getLong(column);
- }
+ static OrcValueWriter<?> map(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter,
+ List<TypeDescription> orcType) {
Review comment:
Why need the `orcType` in this map constructor ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szlta commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
szlta commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r726977756
##########
File path: data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriters.java
##########
@@ -531,6 +534,46 @@ public void nonNullWrite(int rowId, Map<K, V> map, ColumnVector output) {
}
}
+ public abstract static class StructWriter<S> implements OrcValueWriter<S> {
+ private final List<OrcValueWriter<?>> writers;
+
+ protected StructWriter(List<OrcValueWriter<?>> writers) {
+ this.writers = writers;
+ }
+
+ public List<OrcValueWriter<?>> writers() {
Review comment:
I think we should leave that as it will come handy for https://github.com/apache/iceberg/pull/3250
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r725863971
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcWriter.java
##########
@@ -126,30 +127,106 @@ public SparkOrcValueWriter primitive(Type.PrimitiveType iPrimitive, TypeDescript
}
}
- private static class StructWriter implements SparkOrcValueWriter {
- private final List<SparkOrcValueWriter> writers;
+ private static class InternalRowWriter extends GenericOrcWriter.StructWriter<InternalRow> {
+ private final List<FieldGetter<?>> fieldGetters;
- StructWriter(List<SparkOrcValueWriter> writers) {
- this.writers = writers;
- }
+ InternalRowWriter(List<OrcValueWriter<?>> writers, List<TypeDescription> orcTypes) {
+ super(writers);
+ this.fieldGetters = Lists.newArrayListWithExpectedSize(orcTypes.size());
- List<SparkOrcValueWriter> writers() {
- return writers;
+ for (int i = 0; i < orcTypes.size(); i++) {
+ fieldGetters.add(createFieldGetter(orcTypes.get(i)));
+ }
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- InternalRow value = data.getStruct(column, writers.size());
- StructColumnVector cv = (StructColumnVector) output;
- for (int c = 0; c < writers.size(); ++c) {
- writers.get(c).write(rowId, c, value, cv.fields[c]);
- }
+ public Class<?> getJavaClass() {
+ return InternalRow.class;
}
@Override
public Stream<FieldMetrics<?>> metrics() {
- return writers.stream().flatMap(SparkOrcValueWriter::metrics);
+ return writers().stream().flatMap(OrcValueWriter::metrics);
+ }
+
+ @Override
+ protected Object get(InternalRow struct, int index) {
+ return fieldGetters.get(index).getFieldOrNull(struct, index);
+ }
+ }
+
+ static FieldGetter<?> createFieldGetter(TypeDescription fieldType) {
+ final FieldGetter<?> fieldGetter;
+ switch (fieldType.getCategory()) {
+ case BOOLEAN:
+ fieldGetter = (row, ordinal) -> row.getBoolean(ordinal);
Review comment:
Nit: This lambda could be replaced with method reference :
```
SpecializedGetters::getBoolean
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r727058479
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -267,28 +152,39 @@ public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnV
growColumnVector(cv.child, cv.childCount);
// Add each element
for (int e = 0; e < cv.lengths[rowId]; ++e) {
- writer.write((int) (e + cv.offsets[rowId]), e, value, cv.child);
+ writer.write((int) (e + cv.offsets[rowId]), fieldGetter.getFieldOrNull(value, e), cv.child);
}
}
@Override
public Stream<FieldMetrics<?>> metrics() {
return writer.metrics();
}
+
+ @Override
+ public Class<?> getJavaClass() {
+ return ArrayData.class;
+ }
}
- private static class MapWriter implements SparkOrcValueWriter {
- private final SparkOrcValueWriter keyWriter;
- private final SparkOrcValueWriter valueWriter;
+ private static class MapWriter implements OrcValueWriter<MapData> {
+ private final OrcValueWriter keyWriter;
+ private final OrcValueWriter valueWriter;
+ private final SparkOrcWriter.FieldGetter keyFieldGetter;
+ private final SparkOrcWriter.FieldGetter valueFieldGetter;
- MapWriter(SparkOrcValueWriter keyWriter, SparkOrcValueWriter valueWriter) {
+ MapWriter(OrcValueWriter<?> keyWriter, OrcValueWriter<?> valueWriter, List<TypeDescription> orcTypes) {
Review comment:
Nit: same comment as the https://github.com/apache/iceberg/pull/3248/files#r727053434
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: ORC: Refactor value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r727064747
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueWriters.java
##########
@@ -267,28 +152,39 @@ public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnV
growColumnVector(cv.child, cv.childCount);
// Add each element
for (int e = 0; e < cv.lengths[rowId]; ++e) {
- writer.write((int) (e + cv.offsets[rowId]), e, value, cv.child);
+ writer.write((int) (e + cv.offsets[rowId]), fieldGetter.getFieldOrNull(value, e), cv.child);
}
}
@Override
public Stream<FieldMetrics<?>> metrics() {
return writer.metrics();
}
+
+ @Override
+ public Class<?> getJavaClass() {
+ return ArrayData.class;
+ }
}
- private static class MapWriter implements SparkOrcValueWriter {
- private final SparkOrcValueWriter keyWriter;
- private final SparkOrcValueWriter valueWriter;
+ private static class MapWriter implements OrcValueWriter<MapData> {
+ private final OrcValueWriter keyWriter;
+ private final OrcValueWriter valueWriter;
+ private final SparkOrcWriter.FieldGetter keyFieldGetter;
+ private final SparkOrcWriter.FieldGetter valueFieldGetter;
Review comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r725859974
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcWriter.java
##########
@@ -42,75 +46,72 @@
*/
public class SparkOrcWriter implements OrcRowWriter<InternalRow> {
- private final SparkOrcValueWriter writer;
+ private final InternalRowWriter writer;
public SparkOrcWriter(Schema iSchema, TypeDescription orcSchema) {
Preconditions.checkArgument(orcSchema.getCategory() == TypeDescription.Category.STRUCT,
"Top level must be a struct " + orcSchema);
- writer = OrcSchemaWithTypeVisitor.visit(iSchema, orcSchema, new WriteBuilder());
+ writer = (InternalRowWriter) OrcSchemaWithTypeVisitor.visit(iSchema, orcSchema, new WriteBuilder());
}
@Override
public void write(InternalRow value, VectorizedRowBatch output) {
- Preconditions.checkArgument(writer instanceof StructWriter, "writer must be StructWriter");
+ Preconditions.checkArgument(value != null, "value must not be null");
Review comment:
Why does the value not allow to be null ? In my view, it's possible for people to append a `null` row to orc file..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #3248: Refactor ORC value writers
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #3248:
URL: https://github.com/apache/iceberg/pull/3248#discussion_r725870777
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcWriter.java
##########
@@ -126,30 +127,106 @@ public SparkOrcValueWriter primitive(Type.PrimitiveType iPrimitive, TypeDescript
}
}
- private static class StructWriter implements SparkOrcValueWriter {
- private final List<SparkOrcValueWriter> writers;
+ private static class InternalRowWriter extends GenericOrcWriter.StructWriter<InternalRow> {
+ private final List<FieldGetter<?>> fieldGetters;
- StructWriter(List<SparkOrcValueWriter> writers) {
- this.writers = writers;
- }
+ InternalRowWriter(List<OrcValueWriter<?>> writers, List<TypeDescription> orcTypes) {
+ super(writers);
+ this.fieldGetters = Lists.newArrayListWithExpectedSize(orcTypes.size());
- List<SparkOrcValueWriter> writers() {
- return writers;
+ for (int i = 0; i < orcTypes.size(); i++) {
+ fieldGetters.add(createFieldGetter(orcTypes.get(i)));
+ }
}
@Override
- public void nonNullWrite(int rowId, int column, SpecializedGetters data, ColumnVector output) {
- InternalRow value = data.getStruct(column, writers.size());
- StructColumnVector cv = (StructColumnVector) output;
- for (int c = 0; c < writers.size(); ++c) {
- writers.get(c).write(rowId, c, value, cv.fields[c]);
- }
+ public Class<?> getJavaClass() {
+ return InternalRow.class;
}
@Override
public Stream<FieldMetrics<?>> metrics() {
- return writers.stream().flatMap(SparkOrcValueWriter::metrics);
+ return writers().stream().flatMap(OrcValueWriter::metrics);
+ }
+
+ @Override
+ protected Object get(InternalRow struct, int index) {
+ return fieldGetters.get(index).getFieldOrNull(struct, index);
+ }
+ }
+
+ static FieldGetter<?> createFieldGetter(TypeDescription fieldType) {
+ final FieldGetter<?> fieldGetter;
+ switch (fieldType.getCategory()) {
+ case BOOLEAN:
+ fieldGetter = (row, ordinal) -> row.getBoolean(ordinal);
+ break;
+ case BYTE:
+ fieldGetter = (row, ordinal) -> row.getByte(ordinal);
+ break;
+ case SHORT:
+ fieldGetter = (row, ordinal) -> row.getShort(ordinal);
+ break;
+ case DATE:
+ case INT:
+ fieldGetter = (row, ordinal) -> row.getInt(ordinal);
+ break;
+ case LONG:
+ case TIMESTAMP:
+ case TIMESTAMP_INSTANT:
+ fieldGetter = (row, ordinal) -> row.getLong(ordinal);
+ break;
+ case FLOAT:
+ fieldGetter = (row, ordinal) -> row.getFloat(ordinal);
+ break;
+ case DOUBLE:
+ fieldGetter = (row, ordinal) -> row.getDouble(ordinal);
+ break;
+ case BINARY:
+ fieldGetter = (row, ordinal) -> row.getBinary(ordinal);
+ // getBinary always makes a copy, so we don't need to worry about it
+ // being changed behind our back.
+ break;
+ case DECIMAL:
+ fieldGetter = (row, ordinal) ->
+ row.getDecimal(ordinal, fieldType.getPrecision(), fieldType.getScale());
+ break;
+ case STRING:
+ case CHAR:
+ case VARCHAR:
+ fieldGetter = (row, ordinal) -> row.getUTF8String(ordinal);
+ break;
+ case STRUCT:
+ fieldGetter = (row, ordinal) -> row.getStruct(ordinal, fieldType.getChildren().size());
+ break;
+ case LIST:
+ fieldGetter = (row, ordinal) -> row.getArray(ordinal);
+ break;
+ case MAP:
+ fieldGetter = (row, ordinal) -> row.getMap(ordinal);
+ break;
+ default:
+ throw new IllegalArgumentException();
Review comment:
A more clear message is necessary for this exception I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org