public static Object getTypeValue(Type type, Object value)
{
    Object toEncode;
    if (Types.isArrayType(type)) {
        throw new UnsupportedOperationException("Unsupported type " + type);
    }
    else if (Types.isMapType(type)) {
        throw new UnsupportedOperationException("Unsupported type " + type);
    }
    else if (type.equals(VARBINARY)) {
        return ((Slice) value).getBytes();
    }
    else if (type instanceof VarcharType) {
        return ((Slice) value).toStringUtf8();
    }
    else {
        return value;
    }
}
 

/**
 * @param type Presto type
 * @param block Block to decode
 * @param position Position in the block to get
 * @return Java object from the Block
 */
private static Object readObject(Type type, Block block, int position)
{
    if (Types.isArrayType(type)) {
        Type elementType = Types.getElementType(type);
        return getArrayFromBlock(elementType, block.getObject(position, Block.class));
    }
    else if (Types.isMapType(type)) {
        return getMapFromBlock(type, block.getObject(position, Block.class));
    }
    else {
        if (type.getJavaType() == Slice.class) {
            Slice slice = (Slice) TypeUtils.readNativeValue(type, block, position);
            return type.equals(VarcharType.VARCHAR) ? slice.toStringUtf8() : slice.getBytes();
        }

        return TypeUtils.readNativeValue(type, block, position);
    }
}
 

/**
 * Overrides original implementation because of usage of 'extra' column.
 */
@Test
@Override
public void testDescribeTable() {
    MaterializedResult actualColumns = this.computeActual("DESC ORDERS").toTestTypes();
    MaterializedResult.Builder builder = MaterializedResult.resultBuilder(this.getQueryRunner().getDefaultSession(), VarcharType.VARCHAR, VarcharType.VARCHAR, VarcharType.VARCHAR, VarcharType.VARCHAR);
    for (MaterializedRow row: actualColumns.getMaterializedRows()) {
        builder.row(row.getField(0), row.getField(1), "", "");
    }
    MaterializedResult filteredActual = builder.build();
    builder = MaterializedResult.resultBuilder(this.getQueryRunner().getDefaultSession(), VarcharType.VARCHAR, VarcharType.VARCHAR, VarcharType.VARCHAR, VarcharType.VARCHAR);
    MaterializedResult expectedColumns = builder
            .row("orderkey", "bigint", "", "")
            .row("custkey", "bigint", "", "")
            .row("orderstatus", "varchar", "", "")
            .row("totalprice", "double", "", "")
            .row("orderdate", "varchar", "", "")
            .row("orderpriority", "varchar", "", "")
            .row("clerk", "varchar", "", "")
            .row("shippriority", "integer", "", "")
            .row("comment", "varchar", "", "").build();
    Assert.assertEquals(filteredActual, expectedColumns, String.format("%s != %s", filteredActual, expectedColumns));
}
 

@Test
public void testNonExistent()
        throws Exception
{
    byte[] json = "{}".getBytes(StandardCharsets.UTF_8);

    JsonKinesisRowDecoder rowDecoder = new JsonKinesisRowDecoder(PROVIDER.get());
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", VarcharType.VARCHAR, "very/deep/varchar", null, null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", BigintType.BIGINT, "no_bigint", null, null, false, false);
    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "row3", DoubleType.DOUBLE, "double/is_missing", null, null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "row4", BooleanType.BOOLEAN, "hello", null, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(json, providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    DecoderTestUtil.checkIsNull(providers, row1);
    DecoderTestUtil.checkIsNull(providers, row2);
    DecoderTestUtil.checkIsNull(providers, row3);
    DecoderTestUtil.checkIsNull(providers, row4);
}
 

@Test
public void testStringNumber()
        throws Exception
{
    byte[] json = "{\"a_number\":481516,\"a_string\":\"2342\"}".getBytes(StandardCharsets.UTF_8);

    JsonKinesisRowDecoder rowDecoder = new JsonKinesisRowDecoder(PROVIDER.get());
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", VarcharType.VARCHAR, "a_number", null, null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", BigintType.BIGINT, "a_number", null, null, false, false);
    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "row3", VarcharType.VARCHAR, "a_string", null, null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "row4", BigintType.BIGINT, "a_string", null, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(json, providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    DecoderTestUtil.checkValue(providers, row1, "481516");
    DecoderTestUtil.checkValue(providers, row2, 481516);
    DecoderTestUtil.checkValue(providers, row3, "2342");
    DecoderTestUtil.checkValue(providers, row4, 2342);
}
 

@Test
public void testFixedWithString()
{
    String str = "Ich bin zwei Oeltanks";
    byte[] row = str.getBytes(StandardCharsets.UTF_8);

    RawKinesisRowDecoder rowDecoder = new RawKinesisRowDecoder();
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", VarcharType.VARCHAR, null, null, null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", VarcharType.VARCHAR, "0", null, null, false, false);
    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "row3", VarcharType.VARCHAR, "0:4", null, null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "row4", VarcharType.VARCHAR, "5:8", null, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(row, providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    DecoderTestUtil.checkValue(providers, row1, str);
    DecoderTestUtil.checkValue(providers, row2, str);
    // these only work for single byte encodings...
    DecoderTestUtil.checkValue(providers, row3, str.substring(0, 4));
    DecoderTestUtil.checkValue(providers, row4, str.substring(5, 8));
}
 

public void handleStringMap(Map<String, String>  map, String prefix, Map<String, Object> data) {
    if (map != null) {
        for (String key : map.keySet()) {
            String columnName = (prefix + key).toLowerCase();
            data.put(columnName, map.get(key));
            if (!getKubeColumn().containsKey(columnName)) {
                UpdateColumns(columnName, new KubeColumn<V1Node>(columnName, VarcharType.createUnboundedVarcharType(), key));
            }
        }
    }
}
 

@Override
public Slice getSlice(int field)
{
    byte[] bytes = getValue(field);
    Type type = getType(field);
    if (type instanceof VarbinaryType) {
        return Slices.wrappedBuffer(bytes);
    }
    else if (type instanceof VarcharType) {
        return Slices.utf8Slice(new String(bytes, UTF_8));
    }
    else {
        throw new PrestoException(NOT_SUPPORTED, "Unsupported type " + type);
    }
}
 

public static org.apache.kudu.Type toKuduClientType(Type type) {
    if (type instanceof VarcharType) {
        return org.apache.kudu.Type.STRING;
    } else if (type == TimestampType.TIMESTAMP) {
        return org.apache.kudu.Type.UNIXTIME_MICROS;
    } else if (type == BigintType.BIGINT) {
        return org.apache.kudu.Type.INT64;
    } else if (type == IntegerType.INTEGER) {
        return org.apache.kudu.Type.INT32;
    } else if (type == SmallintType.SMALLINT) {
        return org.apache.kudu.Type.INT16;
    } else if (type == TinyintType.TINYINT) {
        return org.apache.kudu.Type.INT8;
    } else if (type == RealType.REAL) {
        return org.apache.kudu.Type.FLOAT;
    } else if (type == DoubleType.DOUBLE) {
        return org.apache.kudu.Type.DOUBLE;
    } else if (type == BooleanType.BOOLEAN) {
        return org.apache.kudu.Type.BOOL;
    } else if (type instanceof VarbinaryType) {
        return org.apache.kudu.Type.BINARY;
    } else if (type instanceof DecimalType) {
        return org.apache.kudu.Type.DECIMAL;
    } else if (type == DateType.DATE) {
        return org.apache.kudu.Type.STRING;
    } else if (type instanceof CharType) {
        return org.apache.kudu.Type.STRING;
    } else {
        throw new IllegalStateException("Type mapping implemented for Presto type: " + type);
    }
}
 

private static Type fromKuduClientType(org.apache.kudu.Type ktype, ColumnTypeAttributes attributes) {
    switch (ktype) {
        case STRING:
            return VarcharType.VARCHAR;
        case UNIXTIME_MICROS:
            return TimestampType.TIMESTAMP;
        case INT64:
            return BigintType.BIGINT;
        case INT32:
            return IntegerType.INTEGER;
        case INT16:
            return SmallintType.SMALLINT;
        case INT8:
            return TinyintType.TINYINT;
        case FLOAT:
            return RealType.REAL;
        case DOUBLE:
            return DoubleType.DOUBLE;
        case BOOL:
            return BooleanType.BOOLEAN;
        case BINARY:
            return VarbinaryType.VARBINARY;
        case DECIMAL:
            return DecimalType.createDecimalType(attributes.getPrecision(), attributes.getScale());
        default:
            throw new IllegalStateException("Kudu type not implemented for " + ktype);
    }
}
 

public static Type mappedType(Type sourceType) {
    if (sourceType == DateType.DATE) {
        return VarcharType.VARCHAR;
    } else {
        return sourceType;
    }
}
 

public static NullableValue getColumnValue(Type type, PartialRow row, int i) {
    if (row.isNull(i)) {
        return NullableValue.asNull(type);
    } else {
        if (type instanceof VarcharType) {
            return NullableValue.of(type, utf8Slice(row.getString(i)));
        } else if (type == TimestampType.TIMESTAMP) {
            return NullableValue.of(type, row.getLong(i) / 1000);
        } else if (type == BigintType.BIGINT) {
            return NullableValue.of(type, row.getLong(i));
        } else if (type == IntegerType.INTEGER) {
            return NullableValue.of(type, row.getInt(i));
        } else if (type == SmallintType.SMALLINT) {
            return NullableValue.of(type, row.getShort(i));
        } else if (type == TinyintType.TINYINT) {
            return NullableValue.of(type, row.getByte(i));
        } else if (type == DoubleType.DOUBLE) {
            return NullableValue.of(type, row.getDouble(i));
        } else if (type == RealType.REAL) {
            return NullableValue.of(type, (long) floatToRawIntBits(row.getFloat(i)));
        } else if (type == BooleanType.BOOLEAN) {
            return NullableValue.of(type, row.getBoolean(i));
        } else if (type instanceof VarbinaryType) {
            return NullableValue.of(type, wrappedBuffer(row.getBinary(i)));
        } else if (type instanceof DecimalType) {
            return NullableValue.of(type, row.getDecimal(i));
        } else {
            throw new IllegalStateException("Handling of type " + type + " is not implemented");
        }
    }
}
 

public static Object getJavaValue(Type type, Object nativeValue) {
    if (type instanceof VarcharType) {
        return ((Slice) nativeValue).toStringUtf8();
    } else if (type == TimestampType.TIMESTAMP) {
        return ((Long) nativeValue) * 1000;
    } else if (type == BigintType.BIGINT) {
        return nativeValue;
    } else if (type == IntegerType.INTEGER) {
        return ((Long) nativeValue).intValue();
    } else if (type == SmallintType.SMALLINT) {
        return ((Long) nativeValue).shortValue();
    } else if (type == TinyintType.TINYINT) {
        return ((Long) nativeValue).byteValue();
    } else if (type == DoubleType.DOUBLE) {
        return nativeValue;
    } else if (type == RealType.REAL) {
        // conversion can result in precision lost
        return intBitsToFloat(((Long) nativeValue).intValue());
    } else if (type == BooleanType.BOOLEAN) {
        return nativeValue;
    } else if (type instanceof VarbinaryType) {
        return ((Slice) nativeValue).toByteBuffer();
    } else if (type instanceof DecimalType) {
        return nativeValue;
    } else {
        throw new IllegalStateException("Back conversion not implemented for " + type);
    }
}
 

public static Object getObject(Type type, RowResult row, int field) {
    if (row.isNull(field)) {
        return null;
    } else {
        if (type instanceof VarcharType) {
            return row.getString(field);
        } else if (type == TimestampType.TIMESTAMP) {
            return row.getLong(field) / 1000;
        } else if (type == BigintType.BIGINT) {
            return row.getLong(field);
        } else if (type == IntegerType.INTEGER) {
            return row.getInt(field);
        } else if (type == SmallintType.SMALLINT) {
            return row.getShort(field);
        } else if (type == TinyintType.TINYINT) {
            return row.getByte(field);
        } else if (type == DoubleType.DOUBLE) {
            return row.getDouble(field);
        } else if (type == RealType.REAL) {
            return row.getFloat(field);
        } else if (type == BooleanType.BOOLEAN) {
            return row.getBoolean(field);
        } else if (type instanceof VarbinaryType) {
            return Slices.wrappedBuffer(row.getBinary(field));
        } else if (type instanceof DecimalType) {
            return row.getDecimal(field);
        } else {
            throw new IllegalStateException("getObject not implemented for " + type);
        }
    }
}
 

public static Slice getSlice(Type type, RowResult row, int field) {
    if (type instanceof VarcharType) {
        return Slices.utf8Slice(row.getString(field));
    } else if (type instanceof VarbinaryType) {
        return Slices.wrappedBuffer(row.getBinary(field));
    } else if (type instanceof DecimalType) {
        BigDecimal dec = row.getDecimal(field);
        return Decimals.encodeScaledValue(dec);
    } else {
        throw new IllegalStateException("getSlice not implemented for " + type);
    }
}
 

public static Map.Entry<SchemaTableName, KinesisStreamDescription> createSimpleJsonStreamDescription(String streamName, SchemaTableName schemaTableName)
{
    // Format: {"id" : 1324, "name" : "some string"}
    ArrayList<KinesisStreamFieldDescription> fieldList = new ArrayList<KinesisStreamFieldDescription>();
    fieldList.add(new KinesisStreamFieldDescription("id", BigintType.BIGINT, "id", "comment", null, null, false));
    fieldList.add(new KinesisStreamFieldDescription("name", VarcharType.VARCHAR, "name", "comment", null, null, false));
    KinesisStreamFieldGroup grp = new KinesisStreamFieldGroup("json", fieldList);

    KinesisStreamDescription desc = new KinesisStreamDescription(schemaTableName.getTableName(), schemaTableName.getSchemaName(), streamName, grp);
    return new AbstractMap.SimpleImmutableEntry<>(schemaTableName, desc);
}
 

@Test
public void testSimple()
        throws Exception
{
    byte[] json = ByteStreams.toByteArray(TestJsonDecoder.class.getResourceAsStream("/decoder/json/message.json"));

    JsonKinesisRowDecoder rowDecoder = new JsonKinesisRowDecoder(PROVIDER.get());
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", VarcharType.VARCHAR, "source", null, null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", VarcharType.VARCHAR, "user/screen_name", null, null, false, false);
    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "row3", BigintType.BIGINT, "id", null, null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "row4", BigintType.BIGINT, "user/statuses_count", null, null, false, false);
    KinesisColumnHandle row5 = new KinesisColumnHandle("", 4, "row5", BooleanType.BOOLEAN, "user/geo_enabled", null, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4, row5);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(json, providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    DecoderTestUtil.checkValue(providers, row1, "<a href=\"http://twitterfeed.com\" rel=\"nofollow\">twitterfeed</a>");
    DecoderTestUtil.checkValue(providers, row2, "EKentuckyNews");
    DecoderTestUtil.checkValue(providers, row3, 493857959588286460L);
    DecoderTestUtil.checkValue(providers, row4, 7630);
    DecoderTestUtil.checkValue(providers, row5, true);
}
 

@Test
public void testOtherExtracts()
        throws Exception
{
    // Test other scenarios: deeper dive into object, get JSON constructs as strings, etc.
    byte[] json = ByteStreams.toByteArray(TestJsonDecoder.class.getResourceAsStream("/decoder/json/event.json"));

    JsonKinesisRowDecoder rowDecoder = new JsonKinesisRowDecoder(PROVIDER.get());
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "event_source", VarcharType.VARCHAR, "source", null, null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "user", VarcharType.VARCHAR, "user/handle", null, null, false, false);
    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "user_string", VarcharType.VARCHAR, "user", null, null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "timestamp", BigintType.BIGINT, "timestamp", null, null, false, false);
    KinesisColumnHandle row5 = new KinesisColumnHandle("", 4, "browser_name", VarcharType.VARCHAR, "environment/browser/name", null, null, false, false);
    KinesisColumnHandle row6 = new KinesisColumnHandle("", 5, "tags_array", VarcharType.VARCHAR, "tags", null, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4, row5, row6);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    log.info("Decoding row from event JSON file");
    boolean valid = rowDecoder.decodeRow(json, providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    DecoderTestUtil.checkValue(providers, row1, "otherworld");
    DecoderTestUtil.checkValue(providers, row2, "joeblow");
    DecoderTestUtil.checkValue(providers, row3, "{\"email\":\"[email protected]\",\"handle\":\"joeblow\"}");
    KinesisFieldValueProvider provider = DecoderTestUtil.findValueProvider(providers, row6);
    assertNotNull(provider);
    log.info(new String(provider.getSlice().getBytes(), StandardCharsets.UTF_8));

    DecoderTestUtil.checkValue(providers, row4, 1450214872847L);
    DecoderTestUtil.checkValue(providers, row5, "Chrome");
    DecoderTestUtil.checkValue(providers, row6, "[\"tag1\",\"tag2\",\"tag3\"]");
    log.info("DONE");
}
 

@Test
public void testNullValues()
        throws Exception
{
    byte[] json = "{}".getBytes(StandardCharsets.UTF_8);

    JsonKinesisRowDecoder rowDecoder = new JsonKinesisRowDecoder(PROVIDER.get());
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", BigintType.BIGINT, "a_number", KinesisFieldDecoder.DEFAULT_FIELD_DECODER_NAME, null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", VarcharType.VARCHAR, "a_string", KinesisFieldDecoder.DEFAULT_FIELD_DECODER_NAME, null, false, false);

    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "row3", BigintType.BIGINT, "a_number", SecondsSinceEpochJsonKinesisFieldDecoder.NAME, null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "row4", BigintType.BIGINT, "a_string", SecondsSinceEpochJsonKinesisFieldDecoder.NAME, null, false, false);

    KinesisColumnHandle row5 = new KinesisColumnHandle("", 4, "row5", VarcharType.VARCHAR, "a_number", SecondsSinceEpochJsonKinesisFieldDecoder.NAME, null, false, false);
    KinesisColumnHandle row6 = new KinesisColumnHandle("", 5, "row6", VarcharType.VARCHAR, "a_string", SecondsSinceEpochJsonKinesisFieldDecoder.NAME, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4, row5, row6);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(json, providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    // sanity checks
    DecoderTestUtil.checkIsNull(providers, row1);
    DecoderTestUtil.checkIsNull(providers, row2);
    DecoderTestUtil.checkIsNull(providers, row3);
    DecoderTestUtil.checkIsNull(providers, row4);
    DecoderTestUtil.checkIsNull(providers, row5);
    DecoderTestUtil.checkIsNull(providers, row6);
}
 

@Test
public void testNullValues()
        throws Exception
{
    byte[] json = "{}".getBytes(StandardCharsets.UTF_8);

    JsonKinesisRowDecoder rowDecoder = new JsonKinesisRowDecoder(PROVIDER.get());
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", BigintType.BIGINT, "a_number", KinesisFieldDecoder.DEFAULT_FIELD_DECODER_NAME, null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", VarcharType.VARCHAR, "a_string", KinesisFieldDecoder.DEFAULT_FIELD_DECODER_NAME, null, false, false);

    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "row3", BigintType.BIGINT, "a_number", MillisecondsSinceEpochJsonKinesisFieldDecoder.NAME, null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "row4", BigintType.BIGINT, "a_string", MillisecondsSinceEpochJsonKinesisFieldDecoder.NAME, null, false, false);

    KinesisColumnHandle row5 = new KinesisColumnHandle("", 4, "row5", VarcharType.VARCHAR, "a_number", MillisecondsSinceEpochJsonKinesisFieldDecoder.NAME, null, false, false);
    KinesisColumnHandle row6 = new KinesisColumnHandle("", 5, "row6", VarcharType.VARCHAR, "a_string", MillisecondsSinceEpochJsonKinesisFieldDecoder.NAME, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4, row5, row6);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(json, providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    // sanity checks
    DecoderTestUtil.checkIsNull(providers, row1);
    DecoderTestUtil.checkIsNull(providers, row2);
    DecoderTestUtil.checkIsNull(providers, row3);
    DecoderTestUtil.checkIsNull(providers, row4);
    DecoderTestUtil.checkIsNull(providers, row5);
    DecoderTestUtil.checkIsNull(providers, row6);
}
 

@Test
public void testNullValues()
        throws Exception
{
    byte[] json = "{}".getBytes(StandardCharsets.UTF_8);

    JsonKinesisRowDecoder rowDecoder = new JsonKinesisRowDecoder(PROVIDER.get());
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", BigintType.BIGINT, "a_number", KinesisFieldDecoder.DEFAULT_FIELD_DECODER_NAME, null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", VarcharType.VARCHAR, "a_string", KinesisFieldDecoder.DEFAULT_FIELD_DECODER_NAME, null, false, false);

    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "row3", BigintType.BIGINT, "a_number", ISO8601JsonKinesisFieldDecoder.NAME, null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "row4", BigintType.BIGINT, "a_string", ISO8601JsonKinesisFieldDecoder.NAME, null, false, false);

    KinesisColumnHandle row5 = new KinesisColumnHandle("", 4, "row5", VarcharType.VARCHAR, "a_number", ISO8601JsonKinesisFieldDecoder.NAME, null, false, false);
    KinesisColumnHandle row6 = new KinesisColumnHandle("", 5, "row6", VarcharType.VARCHAR, "a_string", ISO8601JsonKinesisFieldDecoder.NAME, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4, row5, row6);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(json, providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    // sanity checks
    DecoderTestUtil.checkIsNull(providers, row1);
    DecoderTestUtil.checkIsNull(providers, row2);
    DecoderTestUtil.checkIsNull(providers, row3);
    DecoderTestUtil.checkIsNull(providers, row4);
    DecoderTestUtil.checkIsNull(providers, row5);
    DecoderTestUtil.checkIsNull(providers, row6);
}
 

@Test
public void testNullValues()
        throws Exception
{
    byte[] json = "{}".getBytes(StandardCharsets.UTF_8);

    JsonKinesisRowDecoder rowDecoder = new JsonKinesisRowDecoder(PROVIDER.get());
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", BigintType.BIGINT, "a_number", KinesisFieldDecoder.DEFAULT_FIELD_DECODER_NAME, null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", VarcharType.VARCHAR, "a_string", KinesisFieldDecoder.DEFAULT_FIELD_DECODER_NAME, null, false, false);

    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "row3", BigintType.BIGINT, "a_number", RFC2822JsonKinesisFieldDecoder.NAME, null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "row4", BigintType.BIGINT, "a_string", RFC2822JsonKinesisFieldDecoder.NAME, null, false, false);

    KinesisColumnHandle row5 = new KinesisColumnHandle("", 4, "row5", VarcharType.VARCHAR, "a_number", RFC2822JsonKinesisFieldDecoder.NAME, null, false, false);
    KinesisColumnHandle row6 = new KinesisColumnHandle("", 5, "row6", VarcharType.VARCHAR, "a_string", RFC2822JsonKinesisFieldDecoder.NAME, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4, row5, row6);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(json, providers, columns, map(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    // sanity checks
    checkIsNull(providers, row1);
    checkIsNull(providers, row2);
    checkIsNull(providers, row3);
    checkIsNull(providers, row4);
    checkIsNull(providers, row5);
    checkIsNull(providers, row6);
}
 

@Test
public void testSimple()
{
    ByteBuffer buf = ByteBuffer.allocate(100);
    buf.putLong(4815162342L); // 0 - 7
    buf.putInt(12345678); // 8 - 11
    buf.putShort((short) 4567); // 12 - 13
    buf.put((byte) 123); // 14
    buf.put("Ich bin zwei Oeltanks".getBytes(StandardCharsets.UTF_8)); // 15+

    byte[] row = new byte[buf.position()];
    System.arraycopy(buf.array(), 0, row, 0, buf.position());

    RawKinesisRowDecoder rowDecoder = new RawKinesisRowDecoder();
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", BigintType.BIGINT, "0", "LONG", null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", BigintType.BIGINT, "8", "INT", null, false, false);
    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "row3", BigintType.BIGINT, "12", "SHORT", null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "row4", BigintType.BIGINT, "14", "BYTE", null, false, false);
    KinesisColumnHandle row5 = new KinesisColumnHandle("", 4, "row5", VarcharType.VARCHAR, "15", null, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4, row5);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(row, providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    DecoderTestUtil.checkValue(providers, row1, 4815162342L);
    DecoderTestUtil.checkValue(providers, row2, 12345678);
    DecoderTestUtil.checkValue(providers, row3, 4567);
    DecoderTestUtil.checkValue(providers, row4, 123);
    DecoderTestUtil.checkValue(providers, row5, "Ich bin zwei Oeltanks");
}
 

@SuppressWarnings("NumericCastThatLosesPrecision")
@Test
public void testFloatStuff()
{
    ByteBuffer buf = ByteBuffer.allocate(100);
    buf.putDouble(Math.PI);
    buf.putFloat((float) Math.E);
    buf.putDouble(Math.E);

    byte[] row = new byte[buf.position()];
    System.arraycopy(buf.array(), 0, row, 0, buf.position());

    RawKinesisRowDecoder rowDecoder = new RawKinesisRowDecoder();
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", VarcharType.VARCHAR, null, "DOUBLE", null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", VarcharType.VARCHAR, "8", "FLOAT", null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(row, providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    DecoderTestUtil.checkValue(providers, row1, Math.PI);
    DecoderTestUtil.checkValue(providers, row2, Math.E);
}
 

@Test
public void testSimple()
{
    String csv = "\"row 1\",row2,\"row3\",100,\"200\",300,4.5";

    CsvKinesisRowDecoder rowDecoder = new CsvKinesisRowDecoder();
    KinesisColumnHandle row1 = new KinesisColumnHandle("", 0, "row1", VarcharType.VARCHAR, "0", null, null, false, false);
    KinesisColumnHandle row2 = new KinesisColumnHandle("", 1, "row2", VarcharType.VARCHAR, "1", null, null, false, false);
    KinesisColumnHandle row3 = new KinesisColumnHandle("", 2, "row3", VarcharType.VARCHAR, "2", null, null, false, false);
    KinesisColumnHandle row4 = new KinesisColumnHandle("", 3, "row4", BigintType.BIGINT, "3", null, null, false, false);
    KinesisColumnHandle row5 = new KinesisColumnHandle("", 4, "row5", BigintType.BIGINT, "4", null, null, false, false);
    KinesisColumnHandle row6 = new KinesisColumnHandle("", 5, "row6", BigintType.BIGINT, "5", null, null, false, false);
    KinesisColumnHandle row7 = new KinesisColumnHandle("", 6, "row7", DoubleType.DOUBLE, "6", null, null, false, false);

    List<KinesisColumnHandle> columns = ImmutableList.of(row1, row2, row3, row4, row5, row6, row7);
    Set<KinesisFieldValueProvider> providers = new HashSet<>();

    boolean valid = rowDecoder.decodeRow(csv.getBytes(StandardCharsets.UTF_8), providers, columns, buildMap(columns));
    assertTrue(valid);

    assertEquals(providers.size(), columns.size());

    DecoderTestUtil.checkValue(providers, row1, "row 1");
    DecoderTestUtil.checkValue(providers, row2, "row2");
    DecoderTestUtil.checkValue(providers, row3, "row3");
    DecoderTestUtil.checkValue(providers, row4, 100);
    DecoderTestUtil.checkValue(providers, row5, 200);
    DecoderTestUtil.checkValue(providers, row6, 300);
    DecoderTestUtil.checkValue(providers, row7, 4.5d);
}
 

public Map<String, Object> kubeAPIToData(V1Node node) {
    Map<String, Object> nodeData = new HashMap<String, Object>();
    for (String key : kubeColumn.keySet()) {
        KubeDataGetInterface kubeDataGetInterface = kubeColumn.get(key);
        if (kubeDataGetInterface.getDataSrc() == null) {
            nodeData.put(key, kubeDataGetInterface.getData(node));
        }
    }
    Map<String, String> labels = node.getMetadata().getLabels();
    handleStringMap(labels, NodeLabelsPrefix, nodeData);
    Map<String, String> annotations = node.getMetadata().getAnnotations();
    handleStringMap(annotations, NodeAnnotationsPrefix, nodeData);
    Map<String, Quantity> allocatable = node.getStatus().getAllocatable();
    handleQuantity(allocatable, NodeAllocatablePrefix, nodeData);
    Map<String, Quantity> capacity = node.getStatus().getCapacity();
    handleQuantity(capacity, NodeCapacityPrefix, nodeData);

    List<V1NodeCondition> conditions = node.getStatus().getConditions();
    if (conditions != null) {
        for (V1NodeCondition condition : conditions) {
            String columnNamePrefix = condition.getType().toLowerCase();
            String columnName;
            columnName = columnNamePrefix + "." + ConditonLasttransitiontime;
            nodeData.put(columnName, condition.getLastTransitionTime());
            if (!getKubeColumn().containsKey(columnName)) {
                UpdateColumns(columnName, new KubeColumn<V1Node>(columnName, TimestampType.TIMESTAMP, columnName));
            }
            columnName = columnNamePrefix + "." + ConditionLastheartbeattime;
            nodeData.put(columnName, condition.getLastHeartbeatTime());
            if (!getKubeColumn().containsKey(columnName)) {
                UpdateColumns(columnName, new KubeColumn<V1Node>(columnName, TimestampType.TIMESTAMP, columnName));
            }
            columnName = columnNamePrefix + "." + ConditionStatus;
            nodeData.put(columnName, condition.getStatus());
            if (!getKubeColumn().containsKey(columnName)) {
                UpdateColumns(columnName, new KubeColumn<V1Node>(columnName, VarcharType.createUnboundedVarcharType(), columnName));
            }
            columnName = columnNamePrefix + "." + ConditionMessage;
            nodeData.put(columnName, condition.getMessage());
            if (!getKubeColumn().containsKey(columnName)) {
                UpdateColumns(columnName, new KubeColumn<V1Node>(columnName, VarcharType.createUnboundedVarcharType(), columnName));
            }
            columnName = columnNamePrefix + "." + ConditionReason;
            nodeData.put(columnName, condition.getReason());
            if (!getKubeColumn().containsKey(columnName)) {
                UpdateColumns(columnName, new KubeColumn<V1Node>(columnName, VarcharType.createUnboundedVarcharType(), columnName));
            }
        }
    }
    return nodeData;
}
 

private static Map<String, String> getQueryDsl(TupleDomain<ColumnHandle> constraint)
{
    final Map<String, Object> mergeDslMap = new HashMap<>();
    Map<String, String> dslCacher = new HashMap<>();

    if (constraint.getColumnDomains().isPresent()) {
        for (TupleDomain.ColumnDomain<ColumnHandle> cd : constraint.getColumnDomains().get()) {
            ElasticsearchColumnHandle column = (ElasticsearchColumnHandle) cd.getColumn();
            String columnName = column.getName();

            if ("_type".equals(columnName)) {
                throw new UnsupportedOperationException("this _type filter have't support!");
            }
            else if (columnName.startsWith("_")) {
                getRangesFromDomain(cd.getDomain()).forEach(range -> {
                    checkArgument(range.isSingleValue(), "dsl is must [=] demo where _dsl = \"..dsl string\"");
                    checkArgument(range.getType() instanceof VarcharType, "_dsl filter is not string");
                    String dsl = ((Slice) range.getSingleValue()).toStringUtf8();
                    dslCacher.put(columnName, dsl);
                    if (!"_dsl".equals(columnName)) {
                        dsl = dsl.replace(MatchQueryFunction.MATCH_COLUMN_SEP, columnName.substring(1));
                    }
                    addEsQueryFilter(mergeDslMap, dsl);
                });
            }
            else {
                getRangesFromDomain(cd.getDomain()).forEach(range -> {
                    checkArgument(column.getType().equals(range.getType()), "filter type is " + range.getType() + " but column [" + columnName + "] type is " + column.getType());
                    QueryBuilder queryBuilder = getQueryBuilderFromPrestoRange(columnName, range);
                    addEsQueryFilter(mergeDslMap, queryBuilder.toString());
                });
            }
        }
    }
    try {
        String allDsl = mergeDslMap.isEmpty() ? QueryBuilders.boolQuery().toString() : MAPPER.writeValueAsString(mergeDslMap);
        dslCacher.put("_allDsl", allDsl);
        return dslCacher;
    }
    catch (JsonProcessingException e) {
        throw new PrestoException(ES_DSL_ERROR, e);
    }
}
 

private static Map<String, String> getQueryDsl(TupleDomain<ColumnHandle> constraint)
{
    final Map<String, Object> mergeDslMap = new HashMap<>();
    Map<String, String> dslCacher = new HashMap<>();

    if (constraint.getColumnDomains().isPresent()) {
        for (TupleDomain.ColumnDomain<ColumnHandle> cd : constraint.getColumnDomains().get()) {
            ElasticsearchColumnHandle column = (ElasticsearchColumnHandle) cd.getColumn();
            String columnName = column.getName();

            if ("_type".equals(columnName)) {
                throw new UnsupportedOperationException("this _type filter have't support!");
            }
            else if (columnName.startsWith("_")) {
                getRangesFromDomain(cd.getDomain()).forEach(range -> {
                    checkArgument(range.isSingleValue(), "dsl is must [=] demo where _dsl = \"..dsl string\"");
                    checkArgument(range.getType() instanceof VarcharType, "_dsl filter is not string");
                    String dsl = ((Slice) range.getSingleValue()).toStringUtf8();
                    dslCacher.put(columnName, dsl);
                    if (!"_dsl".equals(columnName)) {
                        dsl = dsl.replace(MatchQueryFunction.MATCH_COLUMN_SEP, columnName.substring(1));
                    }
                    addEsQueryFilter(mergeDslMap, dsl);
                });
            }
            else {
                getRangesFromDomain(cd.getDomain()).forEach(range -> {
                    checkArgument(column.getType().equals(range.getType()), "filter type is " + range.getType() + " but column [" + columnName + "] type is " + column.getType());
                    QueryBuilder queryBuilder = getQueryBuilderFromPrestoRange(columnName, range);
                    addEsQueryFilter(mergeDslMap, queryBuilder.toString());
                });
            }
        }
    }
    try {
        String allDsl = mergeDslMap.isEmpty() ? QueryBuilders.boolQuery().toString() :
                MAPPER.writeValueAsString(mergeDslMap.get("query"));   //es5和 6开始只能返回 query的自节点
        dslCacher.put("_allDsl", allDsl);
        return dslCacher;
    }
    catch (JsonProcessingException e) {
        throw new PrestoException(ES_DSL_ERROR, e);
    }
}
 

private static Map<String, String> getQueryDsl(TupleDomain<ColumnHandle> constraint)
{
    final Map<String, Object> mergeDslMap = new HashMap<>();
    Map<String, String> dslCacher = new HashMap<>();

    if (constraint.getColumnDomains().isPresent()) {
        for (TupleDomain.ColumnDomain<ColumnHandle> cd : constraint.getColumnDomains().get()) {
            ElasticsearchColumnHandle column = (ElasticsearchColumnHandle) cd.getColumn();
            String columnName = column.getName();

            if ("_type".equals(columnName)) {
                throw new UnsupportedOperationException("this _type filter have't support!");
            }
            else if (columnName.startsWith("_")) {
                getRangesFromDomain(cd.getDomain()).forEach(range -> {
                    checkArgument(range.isSingleValue(), "dsl is must [=] demo where _dsl = \"..dsl string\"");
                    checkArgument(range.getType() instanceof VarcharType, "_dsl filter is not string");
                    String dsl = ((Slice) range.getSingleValue()).toStringUtf8();
                    dslCacher.put(columnName, dsl);
                    if (!"_dsl".equals(columnName)) {
                        dsl = dsl.replace(MatchQueryFunction.MATCH_COLUMN_SEP, columnName.substring(1));
                    }
                    addEsQueryFilter(mergeDslMap, dsl);
                });
            }
            else {
                getRangesFromDomain(cd.getDomain()).forEach(range -> {
                    checkArgument(column.getType().equals(range.getType()), "filter type is " + range.getType() + " but column [" + columnName + "] type is " + column.getType());
                    QueryBuilder queryBuilder = getQueryBuilderFromPrestoRange(columnName, range);
                    addEsQueryFilter(mergeDslMap, queryBuilder.toString());
                });
            }
        }
    }
    try {
        String allDsl = mergeDslMap.isEmpty() ? QueryBuilders.boolQuery().toString() :
                MAPPER.writeValueAsString(mergeDslMap.get("query"));   //es5和 6开始只能返回 query的自节点
        dslCacher.put("_allDsl", allDsl);
        return dslCacher;
    }
    catch (JsonProcessingException e) {
        throw new PrestoException(ES_DSL_ERROR, e);
    }
}
 

private Type getType(String typeName)
{
    log.debug("Get type " + typeName);
    typeName = typeName.toLowerCase();
    // check if type is varchar(xx)
    Pattern vcpattern = Pattern.compile("varchar\\(\\s*(\\d+)\\s*\\)");
    Matcher vcmatcher = vcpattern.matcher(typeName);
    if (vcmatcher.find()) {
        String vlen = vcmatcher.group(1);
        if (!vlen.isEmpty()) {
            return VarcharType.createVarcharType(Integer.parseInt(vlen));
        }
        return UnknownType.UNKNOWN;
    }
    // check if type is char(xx)
    Pattern cpattern = Pattern.compile("char\\(\\s*(\\d+)\\s*\\)");
    Matcher cmatcher = cpattern.matcher(typeName);
    if (cmatcher.find()) {
        String clen = cmatcher.group(1);
        if (!clen.isEmpty()) {
            return CharType.createCharType(Integer.parseInt(clen));
        }
        return UnknownType.UNKNOWN;
    }
    // check if type is decimal(precision, scale)
    Pattern dpattern = Pattern.compile("decimal\\((\\d+)\\s*,?\\s*(\\d*)\\)");
    Matcher dmatcher = dpattern.matcher(typeName);
    if (dmatcher.find()) {
        String dprecision = dmatcher.group(1);
        String dscale = dmatcher.group(2);
        if (dprecision.isEmpty()) {
            return UnknownType.UNKNOWN;
        }
        if (dscale.isEmpty()) {
            return DecimalType.createDecimalType(Integer.parseInt(dprecision));
        }
        return DecimalType.createDecimalType(Integer.parseInt(dprecision), Integer.parseInt(dscale));
    }
    switch (typeName) {
        case "boolean":
            return BooleanType.BOOLEAN;
        case "tinyint":
            return TinyintType.TINYINT;
        case "smallint":
            return SmallintType.SMALLINT;
        case "integer":
            return IntegerType.INTEGER;
        case "bigint":
            return BigintType.BIGINT;
        case "real":
            return RealType.REAL;
        case "double":
            return DoubleType.DOUBLE;
        case "date":
            return DateType.DATE;
        case "time":
            return TimeType.TIME;
        case "timestamp":
            return TimestampType.TIMESTAMP;
        default:
            return UnknownType.UNKNOWN;
    }
}