You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@orc.apache.org by do...@apache.org on 2022/02/16 00:50:32 UTC
[orc] branch branch-1.7 updated: ORC-236: Support `UNION` type in Java Convert tool (#1025)
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-1.7
in repository https://gitbox.apache.org/repos/asf/orc.git
The following commit(s) were added to refs/heads/branch-1.7 by this push:
new 2f30633 ORC-236: Support `UNION` type in Java Convert tool (#1025)
2f30633 is described below
commit 2f30633759ad1a192db8ccf09957f4f74b316be5
Author: rizaon <ri...@cloudera.com>
AuthorDate: Tue Feb 15 16:47:45 2022 -0800
ORC-236: Support `UNION` type in Java Convert tool (#1025)
### What changes were proposed in this pull request?
This patch add support to convert json to UNION type in orc file. For
example, for schema `struct<foo:uniontype<int,string>>`, the following
json lines can be parsed into UNION type.
```
{"foo": {"tag": 0, "value": 1}}
{"foo": {"tag": 1, "value": "testing"}}
{"foo": {"tag": 0, "value": 3}}
```
### Why are the changes needed?
This add a missing support for UNION type in java convert tool.
### How was this patch tested?
Manually test against handcrafted json file.
(cherry picked from commit 6f4481565761400587c7a266ca4c0a4244c0aa34)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../org/apache/orc/tools/convert/ConvertTool.java | 15 +++++++-
.../org/apache/orc/tools/convert/JsonReader.java | 43 ++++++++++++++++++++++
.../apache/orc/tools/convert/TestJsonReader.java | 31 ++++++++++++++++
3 files changed, 88 insertions(+), 1 deletion(-)
diff --git a/java/tools/src/java/org/apache/orc/tools/convert/ConvertTool.java b/java/tools/src/java/org/apache/orc/tools/convert/ConvertTool.java
index 87a7015..19bead3 100644
--- a/java/tools/src/java/org/apache/orc/tools/convert/ConvertTool.java
+++ b/java/tools/src/java/org/apache/orc/tools/convert/ConvertTool.java
@@ -59,6 +59,8 @@ public class ConvertTool {
private final String csvNullString;
private final String timestampFormat;
private final String bloomFilterColumns;
+ private final String unionTag;
+ private final String unionValue;
private final Writer writer;
private final VectorizedRowBatch batch;
@@ -151,7 +153,8 @@ public class ConvertTool {
}
case JSON: {
FSDataInputStream underlying = filesystem.open(path);
- return new JsonReader(getReader(underlying), underlying, size, schema, timestampFormat);
+ return new JsonReader(getReader(underlying), underlying, size, schema, timestampFormat,
+ unionTag, unionValue);
}
case CSV: {
FSDataInputStream underlying = filesystem.open(path);
@@ -196,6 +199,8 @@ public class ConvertTool {
this.csvNullString = opts.getOptionValue('n', "");
this.timestampFormat = opts.getOptionValue("t", DEFAULT_TIMESTAMP_FORMAT);
this.bloomFilterColumns = opts.getOptionValue('b', null);
+ this.unionTag = opts.getOptionValue("union-tag", "tag");
+ this.unionValue = opts.getOptionValue("union-value", "value");
String outFilename = opts.hasOption('o')
? opts.getOptionValue('o') : "output.orc";
boolean overwrite = opts.hasOption('O');
@@ -274,6 +279,14 @@ public class ConvertTool {
Option.builder("O").longOpt("overwrite").desc("Overwrite an existing file")
.build()
);
+ options.addOption(
+ Option.builder().longOpt("union-tag")
+ .desc("JSON key name representing UNION tag. Default to \"tag\".")
+ .hasArg().build());
+ options.addOption(
+ Option.builder().longOpt("union-value")
+ .desc("JSON key name representing UNION value. Default to \"value\".")
+ .hasArg().build());
CommandLine cli = new DefaultParser().parse(options, args);
if (cli.hasOption('h') || cli.getArgs().length == 0) {
HelpFormatter formatter = new HelpFormatter();
diff --git a/java/tools/src/java/org/apache/orc/tools/convert/JsonReader.java b/java/tools/src/java/org/apache/orc/tools/convert/JsonReader.java
index 8119e6b..8055310 100644
--- a/java/tools/src/java/org/apache/orc/tools/convert/JsonReader.java
+++ b/java/tools/src/java/org/apache/orc/tools/convert/JsonReader.java
@@ -34,6 +34,7 @@ import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.UnionColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.orc.RecordReader;
import org.apache.orc.TypeDescription;
@@ -62,6 +63,8 @@ public class JsonReader implements RecordReader {
private final FSDataInputStream input;
private long rowNumber = 0;
private final DateTimeFormatter dateTimeFormatter;
+ private String unionTag = "tag";
+ private String unionValue = "value";
interface JsonConverter {
void convert(JsonElement value, ColumnVector vect, int row);
@@ -293,6 +296,32 @@ public class JsonReader implements RecordReader {
}
}
+ class UnionColumnConverter implements JsonConverter {
+ private JsonConverter[] childConverter;
+
+ UnionColumnConverter(TypeDescription schema) {
+ int size = schema.getChildren().size();
+ childConverter = new JsonConverter[size];
+ for (int i = 0; i < size; i++) {
+ childConverter[i] = createConverter(schema.getChildren().get(i));
+ }
+ }
+
+ @Override
+ public void convert(JsonElement value, ColumnVector vect, int row) {
+ if (value == null || value.isJsonNull()) {
+ vect.noNulls = false;
+ vect.isNull[row] = true;
+ } else {
+ UnionColumnVector vector = (UnionColumnVector) vect;
+ JsonObject obj = value.getAsJsonObject();
+ int tag = obj.get(unionTag).getAsInt();
+ vector.tags[row] = tag;
+ childConverter[tag].convert(obj.get(unionValue), vector.fields[tag], row);
+ }
+ }
+ }
+
JsonConverter createConverter(TypeDescription schema) {
switch (schema.getCategory()) {
case BYTE:
@@ -324,6 +353,8 @@ public class JsonReader implements RecordReader {
return new ListColumnConverter(schema);
case MAP:
return new MapColumnConverter(schema);
+ case UNION:
+ return new UnionColumnConverter(schema);
default:
throw new IllegalArgumentException("Unhandled type " + schema);
}
@@ -333,6 +364,18 @@ public class JsonReader implements RecordReader {
FSDataInputStream underlying,
long size,
TypeDescription schema,
+ String timestampFormat,
+ String unionTag,
+ String unionValue) throws IOException {
+ this(new JsonStreamParser(reader), underlying, size, schema, timestampFormat);
+ this.unionTag = unionTag;
+ this.unionValue = unionValue;
+ }
+
+ public JsonReader(Reader reader,
+ FSDataInputStream underlying,
+ long size,
+ TypeDescription schema,
String timestampFormat) throws IOException {
this(new JsonStreamParser(reader), underlying, size, schema, timestampFormat);
}
diff --git a/java/tools/src/test/org/apache/orc/tools/convert/TestJsonReader.java b/java/tools/src/test/org/apache/orc/tools/convert/TestJsonReader.java
index 3a32c90..ef385db 100644
--- a/java/tools/src/test/org/apache/orc/tools/convert/TestJsonReader.java
+++ b/java/tools/src/test/org/apache/orc/tools/convert/TestJsonReader.java
@@ -18,8 +18,11 @@
package org.apache.orc.tools.convert;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.DateColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.UnionColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.orc.TypeDescription;
import org.junit.jupiter.api.Test;
@@ -143,4 +146,32 @@ public class TestJsonReader {
assertEquals(datetime6.toInstant(), cv.asScratchTimestamp(5).toInstant());
}
+ @Test
+ public void testUnionTypeSupport() throws IOException {
+ String inputString = "{\"foo\": {\"tag\": 0, \"value\": 1}}\n" +
+ "{\"foo\": {\"tag\": 1, \"value\": \"testing\"}}\n" +
+ "{\"foo\": {\"tag\": 0, \"value\": 3}}";
+
+
+ StringReader input = new StringReader(inputString);
+
+ TypeDescription schema = TypeDescription.fromString("struct<foo:uniontype<int,string>>");
+ JsonReader reader = new JsonReader(input, null, 1, schema, "", "tag", "value");
+ VectorizedRowBatch batch = schema.createRowBatch(3);
+ assertTrue(reader.nextBatch(batch));
+ assertEquals(3, batch.size);
+ UnionColumnVector union = (UnionColumnVector) batch.cols[0];
+ LongColumnVector longs = (LongColumnVector) union.fields[0];
+ BytesColumnVector strs = (BytesColumnVector) union.fields[1];
+ assertTrue(union.noNulls);
+ assertFalse(union.isNull[0]);
+ assertEquals(0, union.tags[0]);
+ assertEquals(1, longs.vector[0]);
+ assertFalse(union.isNull[1]);
+ assertEquals(1, union.tags[1]);
+ assertEquals("testing", strs.toString(1));
+ assertFalse(union.isNull[2]);
+ assertEquals(0, union.tags[2]);
+ assertEquals(3, longs.vector[2]);
+ }
}