Skip to content

Rebase type-safe api branch on master #91

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Open
wants to merge 13 commits into
base: master
Choose a base branch
from
Open

Rebase type-safe api branch on master #91

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
0211810
WIP: Rebase on master branch
wmoustafa Nov 30, 2020
cbda011
Eliminate further StdXXX primitive references and fix some tests
wmoustafa Jan 4, 2020
49db345
Remove further StdData references
wmoustafa Jan 4, 2020
934074d
Address review comments
wmoustafa May 10, 2020
a53d7bf
WIP: Rebase on mater - continue
wmoustafa Nov 30, 2020
48c6c29
Introduce type-parameterized array and map APIs; replace wrapper prim…
wmoustafa Dec 17, 2019
cf95f35
Address review comments
wmoustafa May 10, 2020
2bafcce
Fix build errors
wmoustafa Feb 8, 2021
73feee0
Missed conflicts and fix test issues
maluchari Aug 12, 2021
a906f22
Address review comments
maluchari Aug 23, 2021
591b991
Address review comments
maluchari Aug 23, 2021
a5633a3
Merge branch 'linkedin:master' into malini_rebase_api
maluchari Oct 6, 2021
639330d
Merge branch 'linkedin:master' into malini_rebase_api
maluchari Nov 11, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion transportable-udfs-api/src/main/java/com/linkedin/transport/api/StdFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ public interface StdFactory extends Serializable {
/**
* Creates a {@link RowData} whose type is given by the given {@link StdType}.
*
* It is expected that the top-level {@link StdType} is a {@link RowType}.
* It is expected that the top-level {@link StdType} is a {@link com.linkedin.transport.api.types.RowType}.
*
* @param stdType type of the struct to be created
* @return a {@link RowData} with all fields initialized to null
Expand Down
1 change: 0 additions & 1 deletion transportable-udfs-avro/src/main/java/com/linkedin/transport/avro/AvroFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
import com.linkedin.transport.avro.typesystem.AvroTypeFactory;
import com.linkedin.transport.typesystem.AbstractBoundVariables;
import com.linkedin.transport.typesystem.TypeSignature;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
Expand Down
9 changes: 5 additions & 4 deletions transportable-udfs-avro/src/main/java/com/linkedin/transport/avro/AvroWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,11 @@
import com.linkedin.transport.avro.types.AvroMapType;
import com.linkedin.transport.avro.types.AvroStringType;
import com.linkedin.transport.avro.types.AvroRowType;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericEnumSymbol;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.util.Utf8;

Expand Down Expand Up @@ -70,11 +71,11 @@ public static Object createStdData(Object avroData, Schema avroSchema) {
}

public static Object getPlatformData(Object transportData) {
if (transportData instanceof Integer || transportData instanceof Long || transportData instanceof Double ||
transportData instanceof Boolean || transportData instanceof ByteBuffer) {
if (transportData instanceof Integer || transportData instanceof Long || transportData instanceof Double
|| transportData instanceof Boolean || transportData instanceof ByteBuffer) {
return transportData;
} else if (transportData instanceof String) {
return transportData == null? null : new Utf8((String) transportData);
return transportData == null ? null : new Utf8((String) transportData);
} else {
return transportData == null ? null : ((PlatformData) transportData).getUnderlyingData();
}
Expand Down
2 changes: 1 addition & 1 deletion transportable-udfs-avro/src/main/java/com/linkedin/transport/avro/StdUdfWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ protected Object wrap(Object avroObject, Schema inputSchema, Object stdData) {
case BOOLEAN:
return avroObject;
case STRING:
return avroObject == null? null : avroObject.toString();
return avroObject == null ? null : avroObject.toString();
case ARRAY:
case MAP:
case RECORD:
Expand Down
114 changes: 53 additions & 61 deletions transportable-udfs-avro/src/test/java/com/linkedin/transport/avro/TestAvroWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,36 +8,28 @@
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.linkedin.transport.api.data.PlatformData;
import com.linkedin.transport.api.data.StdData;
import com.linkedin.transport.api.types.StdType;
import com.linkedin.transport.avro.data.AvroArray;
import com.linkedin.transport.avro.data.AvroBinary;
import com.linkedin.transport.avro.data.AvroBoolean;
import com.linkedin.transport.avro.data.AvroDouble;
import com.linkedin.transport.avro.data.AvroFloat;
import com.linkedin.transport.avro.data.AvroInteger;
import com.linkedin.transport.avro.data.AvroLong;
import com.linkedin.transport.avro.data.AvroMap;
import com.linkedin.transport.avro.data.AvroString;
import com.linkedin.transport.avro.data.AvroStruct;
import com.linkedin.transport.avro.data.AvroArrayData;
import com.linkedin.transport.avro.data.AvroMapData;
import com.linkedin.transport.avro.data.AvroRowData;
import com.linkedin.transport.avro.types.AvroArrayType;
import com.linkedin.transport.avro.types.AvroBinaryType;
/*import com.linkedin.transport.avro.types.AvroBinaryType;
import com.linkedin.transport.avro.types.AvroBooleanType;
import com.linkedin.transport.avro.types.AvroDoubleType;
import com.linkedin.transport.avro.types.AvroFloatType;
import com.linkedin.transport.avro.types.AvroIntegerType;
import com.linkedin.transport.avro.types.AvroIntegerType;*/
import com.linkedin.transport.avro.types.AvroLongType;
import com.linkedin.transport.avro.types.AvroMapType;
import com.linkedin.transport.avro.types.AvroStringType;
import com.linkedin.transport.avro.types.AvroStructType;
import java.nio.ByteBuffer;
import com.linkedin.transport.avro.types.AvroRowType;
//import com.linkedin.transport.avro.types.AvroStringType;
//import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Map;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.util.Utf8;
//import org.apache.avro.util.Utf8;
import org.testng.annotations.Test;

import static org.testng.Assert.*;
Expand All @@ -55,14 +47,14 @@ private Schema createSchema(String fieldName, String typeName) {
}

private void testSimpleType(String typeName, Class<? extends StdType> expectedAvroTypeClass,
Object testData, Class<? extends StdData> expectedDataClass) {
Object testData, Class<? extends Object> expectedDataClass) {
Schema avroSchema = createSchema(String.format("\"%s\"", typeName));

StdType stdType = AvroWrapper.createStdType(avroSchema);
assertTrue(expectedAvroTypeClass.isAssignableFrom(stdType.getClass()));
assertEquals(avroSchema, stdType.underlyingType());

StdData stdData = AvroWrapper.createStdData(testData, avroSchema);
Object stdData = AvroWrapper.createStdData(testData, avroSchema);
assertNotNull(stdData);
assertTrue(expectedDataClass.isAssignableFrom(stdData.getClass()));
if ("string".equals(typeName)) {
Expand All @@ -73,41 +65,41 @@ private void testSimpleType(String typeName, Class<? extends StdType> expectedAv
}
}

@Test
/* @Test
public void testBooleanType() {
testSimpleType("boolean", AvroBooleanType.class, true, AvroBoolean.class);
testSimpleType("boolean", AvroBooleanType.class, true, Boolean.class);
}

@Test
public void testIntegerType() {
testSimpleType("int", AvroIntegerType.class, 1, AvroInteger.class);
testSimpleType("int", AvroIntegerType.class, 1, Integer.class);
}

@Test
public void testLongType() {
testSimpleType("long", AvroLongType.class, 1L, AvroLong.class);
testSimpleType("long", AvroLongType.class, 1L, Long.class);
}

@Test
public void testFloatType() {
testSimpleType("float", AvroFloatType.class, 1.0f, AvroFloat.class);
testSimpleType("float", AvroFloatType.class, 1.0f, Float.class);
}

@Test
public void testDoubleType() {
testSimpleType("double", AvroDoubleType.class, 1.0, AvroDouble.class);
testSimpleType("double", AvroDoubleType.class, 1.0, Double.class);
}

@Test
public void testStringType() {
testSimpleType("string", AvroStringType.class, new Utf8("foo"), AvroString.class);
testSimpleType("string", AvroStringType.class, "foo", AvroString.class);
testSimpleType("string", AvroStringType.class, new Utf8("foo"), String.class);
testSimpleType("string", AvroStringType.class, "foo", String.class);
}

@Test
public void testBinaryType() {
testSimpleType("bytes", AvroBinaryType.class, ByteBuffer.wrap("bar".getBytes()), AvroBinary.class);
}
// testSimpleType("bytes", AvroBinaryType.class, ByteBuffer.wrap("bar".getBytes()), Binary.class);
}*/

@Test
public void testEnumType() {
Expand All @@ -122,34 +114,34 @@ public void testEnumType() {

GenericRecord record1 = new GenericData.Record(structSchema);
record1.put("field1", "A");
StdData stdEnumData1 = AvroWrapper.createStdData(record1.get("field1"),
Object stdEnumData1 = AvroWrapper.createStdData(record1.get("field1"),
Schema.createEnum("SampleEnum", "", "", Arrays.asList("A", "B")));
assertTrue(stdEnumData1 instanceof AvroString);
assertEquals("A", ((AvroString) stdEnumData1).get());
assertTrue(stdEnumData1 instanceof String);
assertEquals("A", ((String) stdEnumData1));

GenericRecord record2 = new GenericData.Record(structSchema);
record1.put("field1", new GenericData.EnumSymbol(field1, "A"));
StdData stdEnumData2 = AvroWrapper.createStdData(record1.get("field1"),
Object stdEnumData2 = AvroWrapper.createStdData(record1.get("field1"),
Schema.createEnum("SampleEnum", "", "", Arrays.asList("A", "B")));
assertTrue(stdEnumData2 instanceof AvroString);
assertEquals("A", ((AvroString) stdEnumData2).get());
assertTrue(stdEnumData2 instanceof String);
assertEquals("A", ((String) stdEnumData2));
}

@Test
public void testArrayType() {
Schema elementType = createSchema("\"int\"");
Schema arraySchema = Schema.createArray(elementType);

StdType stdArrayType = AvroWrapper.createStdType(arraySchema);
Object stdArrayType = AvroWrapper.createStdType(arraySchema);
assertTrue(stdArrayType instanceof AvroArrayType);
assertEquals(arraySchema, stdArrayType.underlyingType());
assertEquals(arraySchema, ((AvroArrayType) stdArrayType).underlyingType());
assertEquals(elementType, ((AvroArrayType) stdArrayType).elementType().underlyingType());

GenericArray<Integer> value = new GenericData.Array<>(arraySchema, Arrays.asList(1, 2));
StdData stdArrayData = AvroWrapper.createStdData(value, arraySchema);
assertTrue(stdArrayData instanceof AvroArray);
assertEquals(2, ((AvroArray) stdArrayData).size());
assertEquals(value, ((AvroArray) stdArrayData).getUnderlyingData());
Object stdArrayData = AvroWrapper.createStdData(value, arraySchema);
assertTrue(stdArrayData instanceof AvroArrayData);
assertEquals(2, ((AvroArrayData) stdArrayData).size());
assertEquals(value, ((AvroArrayData) stdArrayData).getUnderlyingData());
}

@Test
Expand All @@ -163,10 +155,10 @@ public void testMapType() {
assertEquals(valueType, ((AvroMapType) stdMapType).valueType().underlyingType());

Map<String, Long> value = ImmutableMap.of("foo", 1L, "bar", 2L);
StdData stdMapData = AvroWrapper.createStdData(value, mapSchema);
assertTrue(stdMapData instanceof AvroMap);
assertEquals(2, ((AvroMap) stdMapData).size());
assertEquals(value, ((AvroMap) stdMapData).getUnderlyingData());
Object stdMapData = AvroWrapper.createStdData(value, mapSchema);
assertTrue(stdMapData instanceof AvroMapData);
assertEquals(2, ((AvroMapData) stdMapData).size());
assertEquals(value, ((AvroMapData) stdMapData).getUnderlyingData());
}

@Test
Expand All @@ -179,21 +171,21 @@ public void testRecordType() {
));

StdType stdStructType = AvroWrapper.createStdType(structSchema);
assertTrue(stdStructType instanceof AvroStructType);
assertTrue(stdStructType instanceof AvroRowType);
assertEquals(structSchema, stdStructType.underlyingType());
assertEquals(field1, ((AvroStructType) stdStructType).fieldTypes().get(0).underlyingType());
assertEquals(field2, ((AvroStructType) stdStructType).fieldTypes().get(1).underlyingType());
assertEquals(field1, ((AvroRowType) stdStructType).fieldTypes().get(0).underlyingType());
assertEquals(field2, ((AvroRowType) stdStructType).fieldTypes().get(1).underlyingType());

GenericRecord value = new GenericData.Record(structSchema);
value.put("field1", 1);
value.put("field2", 2.0);
StdData stdStructData = AvroWrapper.createStdData(value, structSchema);
assertTrue(stdStructData instanceof AvroStruct);
AvroStruct avroStruct = (AvroStruct) stdStructData;
Object stdStructData = AvroWrapper.createStdData(value, structSchema);
assertTrue(stdStructData instanceof AvroRowData);
AvroRowData avroStruct = (AvroRowData) stdStructData;
assertEquals(2, avroStruct.fields().size());
assertEquals(value, avroStruct.getUnderlyingData());
assertEquals(1, ((PlatformData) avroStruct.getField("field1")).getUnderlyingData());
assertEquals(2.0, ((PlatformData) avroStruct.getField("field2")).getUnderlyingData());
assertEquals(1, avroStruct.getField("field1"));
assertEquals(2.0, avroStruct.getField("field2"));
}

@Test
Expand All @@ -205,11 +197,11 @@ public void testValidUnionType() {
assertTrue(stdLongType instanceof AvroLongType);
assertEquals(nonNullType, stdLongType.underlyingType());

StdData stdLongData = AvroWrapper.createStdData(1L, unionSchema);
assertTrue(stdLongData instanceof AvroLong);
assertEquals(1L, ((AvroLong) stdLongData).get());
Object stdLongData = AvroWrapper.createStdData(1L, unionSchema);
assertTrue(stdLongData instanceof Long);
assertEquals(1L, stdLongData);

StdData stdNullData = AvroWrapper.createStdData(null, unionSchema);
Object stdNullData = AvroWrapper.createStdData(null, unionSchema);
assertNull(stdNullData);
}

Expand Down Expand Up @@ -242,21 +234,21 @@ public void testStructWithSimpleUnionField() {
GenericRecord record1 = new GenericData.Record(structSchema);
record1.put("field1", 1);
record1.put("field2", 3.0);
AvroStruct avroStruct1 = (AvroStruct) AvroWrapper.createStdData(record1, structSchema);
AvroRowData avroStruct1 = (AvroRowData) AvroWrapper.createStdData(record1, structSchema);
assertEquals(2, avroStruct1.fields().size());
assertEquals(3.0, ((PlatformData) avroStruct1.getField("field2")).getUnderlyingData());
assertEquals(3.0, avroStruct1.getField("field2"));

GenericRecord record2 = new GenericData.Record(structSchema);
record2.put("field1", 1);
record2.put("field2", null);
AvroStruct avroStruct2 = (AvroStruct) AvroWrapper.createStdData(record2, structSchema);
AvroRowData avroStruct2 = (AvroRowData) AvroWrapper.createStdData(record2, structSchema);
assertEquals(2, avroStruct2.fields().size());
assertNull(avroStruct2.getField("field2"));
assertNull(avroStruct2.fields().get(1));

GenericRecord record3 = new GenericData.Record(structSchema);
record3.put("field1", 1);
AvroStruct avroStruct3 = (AvroStruct) AvroWrapper.createStdData(record3, structSchema);
AvroRowData avroStruct3 = (AvroRowData) AvroWrapper.createStdData(record3, structSchema);
assertEquals(2, avroStruct3.fields().size());
assertNull(avroStruct3.getField("field2"));
assertNull(avroStruct3.fields().get(1));
Expand Down
22 changes: 11 additions & 11 deletions ...le-udfs/src/main/java/com/linkedin/transport/examples/NestedMapFromTwoArraysFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,16 +7,16 @@

import com.google.common.collect.ImmutableList;
import com.linkedin.transport.api.StdFactory;
import com.linkedin.transport.api.data.StdArray;
import com.linkedin.transport.api.data.StdMap;
import com.linkedin.transport.api.data.StdStruct;
import com.linkedin.transport.api.data.ArrayData;
import com.linkedin.transport.api.data.MapData;
import com.linkedin.transport.api.data.RowData;
import com.linkedin.transport.api.types.StdType;
import com.linkedin.transport.api.udf.StdUDF1;
import com.linkedin.transport.api.udf.TopLevelStdUDF;
import java.util.List;


public class NestedMapFromTwoArraysFunction extends StdUDF1<StdArray, StdArray> implements TopLevelStdUDF {
public class NestedMapFromTwoArraysFunction extends StdUDF1<ArrayData, ArrayData> implements TopLevelStdUDF {

private StdType _arrayType;
private StdType _mapType;
Expand All @@ -43,31 +43,31 @@ public void init(StdFactory stdFactory) {
}

@Override
public StdArray eval(StdArray a1) {
StdArray result = getStdFactory().createArray(_arrayType);
public ArrayData eval(ArrayData a1) {
ArrayData result = getStdFactory().createArray(_arrayType);

for (int i = 0; i < a1.size(); i++) {
if (a1.get(i) == null) {
return null;
}
StdStruct inputRow = (StdStruct) a1.get(i);
RowData inputRow = (RowData) a1.get(i);

if (inputRow.getField(0) == null || inputRow.getField(1) == null) {
return null;
}
StdArray kValues = (StdArray) inputRow.getField(0);
StdArray vValues = (StdArray) inputRow.getField(1);
ArrayData kValues = (ArrayData) inputRow.getField(0);
ArrayData vValues = (ArrayData) inputRow.getField(1);

if (kValues.size() != vValues.size()) {
return null;
}

StdMap map = getStdFactory().createMap(_mapType);
MapData map = getStdFactory().createMap(_mapType);
for (int j = 0; j < kValues.size(); j++) {
map.put(kValues.get(j), vValues.get(j));
}

StdStruct outputRow = getStdFactory().createStruct(_rowType);
RowData outputRow = getStdFactory().createStruct(_rowType);
outputRow.setField(0, map);

result.add(outputRow);
Expand Down
8 changes: 3 additions & 5 deletions transportable-udfs-hive/src/main/java/com/linkedin/transport/hive/StdUdfWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,6 @@
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.BinaryObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;


/**
* Base class for all Hive Standard UDFs. It provides a standard way of type validation, binding, and output type
Expand Down Expand Up @@ -74,7 +72,7 @@ public ObjectInspector initialize(ObjectInspector[] arguments) {
_stdUdf.init(_stdFactory);
_requiredFilesProcessed = false;
createStdData();
_outputObjectInspector= hiveTypeInference.getOutputDataType();
_outputObjectInspector = hiveTypeInference.getOutputDataType();
return _outputObjectInspector;
}

Expand Down Expand Up @@ -151,8 +149,8 @@ private Object getPlatformData(Object transportData) {
if (transportData == null) {
return null;
} else if (transportData instanceof Integer || transportData instanceof Long || transportData instanceof Boolean
|| transportData instanceof String || transportData instanceof Float || transportData instanceof Double ||
transportData instanceof ByteBuffer) {
|| transportData instanceof String || transportData instanceof Float || transportData instanceof Double
|| transportData instanceof ByteBuffer) {
return HiveWrapper.getPlatformDataForObjectInspector(transportData, _outputObjectInspector);
} else {
return ((PlatformData) transportData).getUnderlyingData();
Expand Down
Empty file removed ...dfs-spark_2.11/src/test/scala/com/linkedin/transport/spark/data/TestSparkPrimitives.scala
View file
Open in desktop
Empty file.
Loading