@Test public void testRowWithLogicalType() throws Exception { Row testRow = new Row(); testRow.add("id", 1); testRow.add("name", "abc"); testRow.add("date", LocalDate.of(2018, 11, 11)); testRow.add("time", null); testRow.add("timestamp", ZonedDateTime.of(2018, 11 , 11 , 11, 11, 11, 0, ZoneId.of("UTC"))); Schema schema = Schema.recordOf("expectedRecord", Schema.Field.of("id", Schema.nullableOf(Schema.of(Schema.Type.INT))), Schema.Field.of("name", Schema.nullableOf(Schema.of(Schema.Type.STRING))), Schema.Field.of("date", Schema.nullableOf(Schema.of(Schema.LogicalType.DATE))), Schema.Field.of("time", Schema.nullableOf( Schema.of(Schema.LogicalType.TIME_MICROS))), Schema.Field.of("timestamp", Schema.nullableOf( Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)))); StructuredRecord expected = StructuredRecord.builder(schema) .set("id", 1) .set("name", "abc") .setDate("date", LocalDate.of(2018, 11, 11)) .setTime("time", null) .setTimestamp("timestamp", ZonedDateTime.of(2018, 11 , 11 , 11, 11, 11, 0, ZoneId.of("UTC"))).build(); RecordConvertor rc = new RecordConvertor(); StructuredRecord actual = rc.decodeRecord(testRow, schema); Assert.assertEquals(expected, actual); }
public StructuredRecord decodeRecord(Row row, Schema schema) throws RecordConvertorException { if (row.getFields().size() == 1) { Object cell = row.getValue(0); if (cell instanceof StructuredRecord) { return (StructuredRecord) cell; } } StructuredRecord.Builder builder = StructuredRecord.builder(schema); List<Schema.Field> fields = schema.getFields(); for (Schema.Field field : fields) { Schema fSchema = field.getSchema(); boolean isNullable = fSchema.isNullable(); String name = field.getName(); Object value = row.getValue(name); try { Object decodedObj = decode(name, value, field.getSchema()); if (decodedObj instanceof LocalDate) { builder.setDate(name, (LocalDate) decodedObj); } else if (decodedObj instanceof LocalTime) { builder.setTime(name, (LocalTime) decodedObj); } else if (decodedObj instanceof ZonedDateTime) { builder.setTimestamp(name, (ZonedDateTime) decodedObj); } else if (decodedObj instanceof BigDecimal) { builder.setDecimal(name, (BigDecimal) decodedObj); } else { builder.set(name, decodedObj); } } catch (UnexpectedFormatException e) { throw new RecordConvertorException( String.format("Field '%s' of type '%s' cannot be set to '%s'. Make sure the value is " + "being set is inline with the specified schema.", name, isNullable ? fSchema.getNonNullable().getDisplayName() : fSchema.getDisplayName(), value == null ? "NULL" : value), e); } } return builder.build(); }
RecordConvertor implements Serializable { public StructuredRecord decodeRecord(Row row, Schema schema) throws RecordConvertorException { if (row.getFields().size() == 1) { Object cell = row.getValue(0); if (cell instanceof StructuredRecord) { return (StructuredRecord) cell; } } StructuredRecord.Builder builder = StructuredRecord.builder(schema); List<Schema.Field> fields = schema.getFields(); for (Schema.Field field : fields) { Schema fSchema = field.getSchema(); boolean isNullable = fSchema.isNullable(); String name = field.getName(); Object value = row.getValue(name); try { Object decodedObj = decode(name, value, field.getSchema()); if (decodedObj instanceof LocalDate) { builder.setDate(name, (LocalDate) decodedObj); } else if (decodedObj instanceof LocalTime) { builder.setTime(name, (LocalTime) decodedObj); } else if (decodedObj instanceof ZonedDateTime) { builder.setTimestamp(name, (ZonedDateTime) decodedObj); } else if (decodedObj instanceof BigDecimal) { builder.setDecimal(name, (BigDecimal) decodedObj); } else { builder.set(name, decodedObj); } } catch (UnexpectedFormatException e) { throw new RecordConvertorException( String.format("Field '%s' of type '%s' cannot be set to '%s'. Make sure the value is " + "being set is inline with the specified schema.", name, isNullable ? fSchema.getNonNullable().getDisplayName() : fSchema.getDisplayName(), value == null ? "NULL" : value), e); } } return builder.build(); } }
RecordConvertor implements Serializable { public StructuredRecord decodeRecord(Row row, Schema schema) throws RecordConvertorException { if (row.getFields().size() == 1) { Object cell = row.getValue(0); if (cell instanceof StructuredRecord) { return (StructuredRecord) cell; } } StructuredRecord.Builder builder = StructuredRecord.builder(schema); List<Schema.Field> fields = schema.getFields(); for (Schema.Field field : fields) { Schema fSchema = field.getSchema(); boolean isNullable = fSchema.isNullable(); String name = field.getName(); Object value = row.getValue(name); try { Object decodedObj = decode(name, value, field.getSchema()); if (decodedObj instanceof LocalDate) { builder.setDate(name, (LocalDate) decodedObj); } else if (decodedObj instanceof LocalTime) { builder.setTime(name, (LocalTime) decodedObj); } else if (decodedObj instanceof ZonedDateTime) { builder.setTimestamp(name, (ZonedDateTime) decodedObj); } else if (decodedObj instanceof BigDecimal) { builder.setDecimal(name, (BigDecimal) decodedObj); } else { builder.set(name, decodedObj); } } catch (UnexpectedFormatException e) { throw new RecordConvertorException( String.format("Field '%s' of type '%s' cannot be set to '%s'. Make sure the value is " + "being set is inline with the specified schema.", name, isNullable ? fSchema.getNonNullable().getDisplayName() : fSchema.getDisplayName(), value == null ? "NULL" : value), e); } } return builder.build(); }  }
RecordConvertor implements Serializable { public StructuredRecord decodeRecord(Row row, Schema schema) throws RecordConvertorException { if (row.getFields().size() == 1) { Object cell = row.getValue(0); if (cell instanceof StructuredRecord) { return (StructuredRecord) cell; } } StructuredRecord.Builder builder = StructuredRecord.builder(schema); List<Schema.Field> fields = schema.getFields(); for (Schema.Field field : fields) { Schema fSchema = field.getSchema(); boolean isNullable = fSchema.isNullable(); String name = field.getName(); Object value = row.getValue(name); try { Object decodedObj = decode(name, value, field.getSchema()); if (decodedObj instanceof LocalDate) { builder.setDate(name, (LocalDate) decodedObj); } else if (decodedObj instanceof LocalTime) { builder.setTime(name, (LocalTime) decodedObj); } else if (decodedObj instanceof ZonedDateTime) { builder.setTimestamp(name, (ZonedDateTime) decodedObj); } else if (decodedObj instanceof BigDecimal) { builder.setDecimal(name, (BigDecimal) decodedObj); } else { builder.set(name, decodedObj); } } catch (UnexpectedFormatException e) { throw new RecordConvertorException( String.format("Field '%s' of type '%s' cannot be set to '%s'. Make sure the value is " + "being set is inline with the specified schema.", name, isNullable ? fSchema.getNonNullable().getDisplayName() : fSchema.getDisplayName(), value == null ? "NULL" : value), e); } } return builder.build(); }  List<StructuredRecord> toStructureRecord(List<Row> rows, Schema schema); StructuredRecord decodeRecord(Row row, Schema schema); }
RecordConvertor implements Serializable { public StructuredRecord decodeRecord(Row row, Schema schema) throws RecordConvertorException { if (row.getFields().size() == 1) { Object cell = row.getValue(0); if (cell instanceof StructuredRecord) { return (StructuredRecord) cell; } } StructuredRecord.Builder builder = StructuredRecord.builder(schema); List<Schema.Field> fields = schema.getFields(); for (Schema.Field field : fields) { Schema fSchema = field.getSchema(); boolean isNullable = fSchema.isNullable(); String name = field.getName(); Object value = row.getValue(name); try { Object decodedObj = decode(name, value, field.getSchema()); if (decodedObj instanceof LocalDate) { builder.setDate(name, (LocalDate) decodedObj); } else if (decodedObj instanceof LocalTime) { builder.setTime(name, (LocalTime) decodedObj); } else if (decodedObj instanceof ZonedDateTime) { builder.setTimestamp(name, (ZonedDateTime) decodedObj); } else if (decodedObj instanceof BigDecimal) { builder.setDecimal(name, (BigDecimal) decodedObj); } else { builder.set(name, decodedObj); } } catch (UnexpectedFormatException e) { throw new RecordConvertorException( String.format("Field '%s' of type '%s' cannot be set to '%s'. Make sure the value is " + "being set is inline with the specified schema.", name, isNullable ? fSchema.getNonNullable().getDisplayName() : fSchema.getDisplayName(), value == null ? "NULL" : value), e); } } return builder.build(); }  List<StructuredRecord> toStructureRecord(List<Row> rows, Schema schema); StructuredRecord decodeRecord(Row row, Schema schema);  }