Skip to content

Commit

Permalink
NIFI-7551 Add support for VARCHAR to Kudu NAR bundle
Browse files Browse the repository at this point in the history 
 - update Kudu dependencies to Kudu 1.12.0
 - add VARCHAR to Kudu Lookup Service and Processor
 - add tests for VARCHAR columns

Signed-off-by: Pierre Villard <pierre.villard.fr@gmail.com>

This closes apache#4347.
  • Loading branch information
@grishick @agentdalecoper
grishick authored and agentdalecoper committed Jun 26, 2020
1 parent 5f0bb7d commit e4fe93c
Show file tree
Hide file tree
Showing 7 changed files with 42 additions and 14 deletions.
2 changes: 1 addition & 1 deletion nifi-nar-bundles/nifi-kudu-bundle/nifi-kudu-controller-service/pom.xml
View file
Expand Up @@ -27,7 +27,7 @@

<properties>
<exclude.tests>None</exclude.tests>
<kudu.version>1.10.0</kudu.version>
<kudu.version>1.12.0</kudu.version>
</properties>
<build>
<extensions>
Expand Down
1 change: 1 addition & 0 deletions ...u-controller-service/src/main/java/org/apache/nifi/controller/kudu/KuduLookupService.java
View file
Expand Up @@ -321,6 +321,7 @@ private RecordSchema kuduSchemaToNiFiSchema(Schema kuduTableSchema, List<String>
break;
case BINARY:
case STRING:
case VARCHAR:
fields.add(new RecordField(cs.getName(), RecordFieldType.STRING.getDataType()));
break;
case DOUBLE:
Expand Down
8 changes: 8 additions & 0 deletions ...troller-service/src/test/java/org/apache/nifi/controller/kudu/ITestKuduLookupService.java
View file
Expand Up @@ -17,6 +17,7 @@
package org.apache.nifi.controller.kudu;

import org.apache.kudu.ColumnSchema;
import org.apache.kudu.ColumnTypeAttributes;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.CreateTableOptions;
Expand Down Expand Up @@ -99,6 +100,9 @@ public void init() throws Exception {
columns.add(new ColumnSchema.ColumnSchemaBuilder("int32", Type.INT32).build());
columns.add(new ColumnSchema.ColumnSchemaBuilder("int64", Type.INT64).build());
columns.add(new ColumnSchema.ColumnSchemaBuilder("unixtime_micros", Type.UNIXTIME_MICROS).build());
columns.add(new ColumnSchema.ColumnSchemaBuilder("varchar_3", Type.VARCHAR).typeAttributes(
new ColumnTypeAttributes.ColumnTypeAttributesBuilder().length(3).build()
).build());
Schema schema = new Schema(columns);

CreateTableOptions opts = new CreateTableOptions().setRangePartitionColumns(Collections.singletonList("string"));
Expand All @@ -120,6 +124,7 @@ public void init() throws Exception {
row.addInt("int32",3);
row.addLong("int64",4L);
row.addTimestamp("unixtime_micros", new Timestamp(nowMillis));
row.addVarchar("varchar_3", "SFO");
session.apply(insert);

insert = table.newInsert();
Expand All @@ -135,6 +140,7 @@ public void init() throws Exception {
row.addInt("int32",13);
row.addLong("int64",14L);
row.addTimestamp("unixtime_micros", new Timestamp(nowMillis+(1000L * 60 * 60 * 24 * 365))); //+ 1 year
row.addVarchar("varchar_3", "SJC");
session.apply(insert);

session.close();
Expand Down Expand Up @@ -198,6 +204,7 @@ public void multi_key() {
map.put("int32",3);
map.put("int64",4L);
map.put("unixtime_micros", new Timestamp(nowMillis));
map.put("varchar_3", "SFO");
Record result = kuduLookupService.lookup(map).get();
validateRow1(result);
}
Expand Down Expand Up @@ -229,6 +236,7 @@ private void validateRow1(Record result){
assertEquals(3, (int)result.getAsInt("int32"));
assertEquals(4L, (long)result.getAsLong("int64"));
assertEquals(new Timestamp(nowMillis), result.getValue("unixtime_micros"));
assertEquals("SFO", result.getValue("varchar_3"));
}

}
2 changes: 1 addition & 1 deletion nifi-nar-bundles/nifi-kudu-bundle/nifi-kudu-processors/pom.xml
View file
Expand Up @@ -26,7 +26,7 @@

<properties>
<exclude.tests>None</exclude.tests>
<kudu.version>1.10.0</kudu.version>
<kudu.version>1.12.0</kudu.version>
</properties>
<build>
<extensions>
Expand Down
3 changes: 3 additions & 0 deletions ...-kudu-processors/src/main/java/org/apache/nifi/processors/kudu/AbstractKuduProcessor.java
View file
Expand Up @@ -349,6 +349,9 @@ protected void buildPartialRow(Schema schema, PartialRow row, Record record, Lis
case DECIMAL:
row.addDecimal(colIdx, new BigDecimal(DataTypeUtils.toString(value, recordFieldName)));
break;
case VARCHAR:
row.addVarchar(colIdx, DataTypeUtils.toString(value, recordFieldName));
break;
default:
throw new IllegalStateException(String.format("unknown column type %s", colType));
}
Expand Down
9 changes: 7 additions & 2 deletions ...-bundle/nifi-kudu-processors/src/test/java/org/apache/nifi/processors/kudu/ITPutKudu.java
View file
Expand Up @@ -17,6 +17,7 @@
package org.apache.nifi.processors.kudu;

import org.apache.kudu.ColumnSchema;
import org.apache.kudu.ColumnTypeAttributes;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.CreateTableOptions;
Expand Down Expand Up @@ -103,6 +104,9 @@ private void createKuduTable() throws KuduException {
List<ColumnSchema> columns = new ArrayList<>();
columns.add(new ColumnSchema.ColumnSchemaBuilder("id", Type.INT32).key(true).build());
columns.add(new ColumnSchema.ColumnSchemaBuilder("stringval", Type.STRING).build());
columns.add(new ColumnSchema.ColumnSchemaBuilder("varcharval", Type.VARCHAR).typeAttributes(
new ColumnTypeAttributes.ColumnTypeAttributesBuilder().length(256).build()
).build());
columns.add(new ColumnSchema.ColumnSchemaBuilder("num32val", Type.INT32).build());
columns.add(new ColumnSchema.ColumnSchemaBuilder("timestampval", Type.UNIXTIME_MICROS).build());
Schema schema = new Schema(columns);
Expand All @@ -115,14 +119,15 @@ private void createRecordReader(int numOfRecord) throws InitializationException
readerFactory = new MockRecordParser();
readerFactory.addSchemaField("id", RecordFieldType.INT);
readerFactory.addSchemaField("stringVal", RecordFieldType.STRING);
readerFactory.addSchemaField("varcharval", RecordFieldType.STRING);
readerFactory.addSchemaField("num32Val", RecordFieldType.INT);
readerFactory.addSchemaField("timestampVal", RecordFieldType.TIMESTAMP);
// Add two extra columns to test handleSchemaDrift = true.
readerFactory.addSchemaField("doubleVal", RecordFieldType.DOUBLE);
readerFactory.addSchemaField("floatVal", RecordFieldType.FLOAT);

for (int i = 0; i < numOfRecord; i++) {
readerFactory.addRecord(i, "val_" + i, 1000 + i, NOW, 100.88 + i, 100.88 + i);
readerFactory.addRecord(i, "val_" + i, "varchar_val_" + i, 1000 + i, NOW, 100.88 + i, 100.88 + i);
}

testRunner.addControllerService("mock-reader-factory", readerFactory);
Expand Down Expand Up @@ -188,7 +193,7 @@ public void testWriteKudu() throws IOException, InitializationException {
KuduTable kuduTable = client.openTable(DEFAULT_TABLE_NAME);

// Verify the extra field was added.
Assert.assertEquals(6, kuduTable.getSchema().getColumnCount());
Assert.assertEquals(7, kuduTable.getSchema().getColumnCount());
Assert.assertTrue(kuduTable.getSchema().hasColumn("doubleval"));
Assert.assertTrue(kuduTable.getSchema().hasColumn("floatval"));

Expand Down
31 changes: 21 additions & 10 deletions ...undle/nifi-kudu-processors/src/test/java/org/apache/nifi/processors/kudu/TestPutKudu.java
View file
Expand Up @@ -417,64 +417,74 @@ public void testUpdateFlowFiles() throws Exception {

@Test
public void testBuildRow() {
buildPartialRow((long) 1, "foo", (short) 10, "id", "id", false);
buildPartialRow((long) 1, "foo", (short) 10, "id", "id", "SFO",false);
}

@Test
public void testBuildPartialRowNullable() {
buildPartialRow((long) 1, null, (short) 10, "id", "id", false);
buildPartialRow((long) 1, null, (short) 10, "id", "id", null, false);
}

@Test(expected = IllegalArgumentException.class)
public void testBuildPartialRowNullPrimaryKey() {
buildPartialRow(null, "foo", (short) 10, "id", "id", false);
buildPartialRow(null, "foo", (short) 10, "id", "id", "SFO", false);
}

@Test(expected = IllegalArgumentException.class)
public void testBuildPartialRowNotNullable() {
buildPartialRow((long) 1, "foo", null, "id", "id", false);
buildPartialRow((long) 1, "foo", null, "id", "id", "SFO",false);
}

@Test
public void testBuildPartialRowLowercaseFields() {
PartialRow row = buildPartialRow((long) 1, "foo", (short) 10, "id", "ID", true);
PartialRow row = buildPartialRow((long) 1, "foo", (short) 10, "id", "ID", "SFO",true);
row.getLong("id");
}

@Test(expected = IllegalArgumentException.class)
public void testBuildPartialRowLowercaseFieldsFalse() {
PartialRow row = buildPartialRow((long) 1, "foo", (short) 10, "id", "ID", false);
PartialRow row = buildPartialRow((long) 1, "foo", (short) 10, "id", "ID", "SFO",false);
row.getLong("id");
}

@Test
public void testBuildPartialRowLowercaseFieldsKuduUpper() {
PartialRow row = buildPartialRow((long) 1, "foo", (short) 10, "ID", "ID", false);
PartialRow row = buildPartialRow((long) 1, "foo", (short) 10, "ID", "ID", "SFO", false);
row.getLong("ID");
}

@Test(expected = IllegalArgumentException.class)
public void testBuildPartialRowLowercaseFieldsKuduUpperFail() {
PartialRow row = buildPartialRow((long) 1, "foo", (short) 10, "ID", "ID", true);
PartialRow row = buildPartialRow((long) 1, "foo", (short) 10, "ID", "ID", "SFO", true);
row.getLong("ID");
}

private PartialRow buildPartialRow(Long id, String name, Short age, String kuduIdName, String recordIdName, Boolean lowercaseFields) {
@Test
public void testBuildPartialRowVarCharTooLong() {
PartialRow row = buildPartialRow((long) 1, "foo", (short) 10, "id", "ID", "San Francisco", true);
Assert.assertEquals("Kudu client should truncate VARCHAR value to expected length", "San", row.getVarchar("airport_code"));
}

private PartialRow buildPartialRow(Long id, String name, Short age, String kuduIdName, String recordIdName, String airport_code, Boolean lowercaseFields) {
final Schema kuduSchema = new Schema(Arrays.asList(
new ColumnSchema.ColumnSchemaBuilder(kuduIdName, Type.INT64).key(true).build(),
new ColumnSchema.ColumnSchemaBuilder("name", Type.STRING).nullable(true).build(),
new ColumnSchema.ColumnSchemaBuilder("age", Type.INT16).nullable(false).build(),
new ColumnSchema.ColumnSchemaBuilder("updated_at", Type.UNIXTIME_MICROS).nullable(false).build(),
new ColumnSchema.ColumnSchemaBuilder("score", Type.DECIMAL).nullable(true).typeAttributes(
new ColumnTypeAttributes.ColumnTypeAttributesBuilder().precision(9).scale(0).build()
).build(),
new ColumnSchema.ColumnSchemaBuilder("airport_code", Type.VARCHAR).nullable(true).typeAttributes(
new ColumnTypeAttributes.ColumnTypeAttributesBuilder().length(3).build()
).build()));

final RecordSchema schema = new SimpleRecordSchema(Arrays.asList(
new RecordField(recordIdName, RecordFieldType.BIGINT.getDataType()),
new RecordField("name", RecordFieldType.STRING.getDataType()),
new RecordField("age", RecordFieldType.SHORT.getDataType()),
new RecordField("updated_at", RecordFieldType.TIMESTAMP.getDataType()),
new RecordField("score", RecordFieldType.LONG.getDataType())));
new RecordField("score", RecordFieldType.LONG.getDataType()),
new RecordField("airport_code", RecordFieldType.STRING.getDataType())));

Map<String, Object> values = new HashMap<>();
PartialRow row = kuduSchema.newPartialRow();
Expand All @@ -483,6 +493,7 @@ private PartialRow buildPartialRow(Long id, String name, Short age, String kuduI
values.put("age", age);
values.put("updated_at", new Timestamp(System.currentTimeMillis()));
values.put("score", 10000L);
values.put("airport_code", airport_code);
processor.buildPartialRow(
kuduSchema,
row,
Expand Down

0 comments on commit e4fe93c

Please sign in to comment.