From 5bbdb2721774639f9f695b81ba8e876315cca9b9 Mon Sep 17 00:00:00 2001 From: Andy Coates Date: Thu, 5 Mar 2020 14:50:41 +0000 Subject: [PATCH 1/5] chore: partial fix to stop ksql trimming trailing zeros from decimals partial fix for: https://github.com/confluentinc/ksql/issues/4710 Remaining work needs a fix in Connect. Then to enable tests in KSQL. --- .../io/confluent/ksql/json/JsonMapper.java | 2 + .../confluent/ksql/json/JsonMapperTest.java | 37 ++- .../json/ValueSpecJsonSerdeSupplier.java | 27 +- .../test/tools/ExpectedRecordComparator.java | 41 ++- .../ksql/test/PlannedTestRewriterTest.java | 8 + .../json/ValueSpecJsonSerdeSupplierTest.java | 102 ++++++ .../tools/ExpectedRecordComparatorTest.java | 102 ++++++ .../6.0.0_1583419431528/plan.json | 147 +++++++++ .../6.0.0_1583419431528/spec.json | 82 +++++ .../6.0.0_1583419431528/topology | 13 + .../query-validation-tests/decimal.json | 312 ++++++++++-------- .../resources/query-validation-tests/sum.json | 4 +- .../ksql/serde/json/JsonSerdeUtils.java | 5 +- .../ksql/serde/json/KsqlJsonDeserializer.java | 4 +- .../serde/json/KsqlJsonDeserializerTest.java | 44 ++- 15 files changed, 764 insertions(+), 166 deletions(-) create mode 100644 ksql-functional-tests/src/test/java/io/confluent/ksql/test/serde/json/ValueSpecJsonSerdeSupplierTest.java create mode 100644 ksql-functional-tests/src/test/java/io/confluent/ksql/test/tools/ExpectedRecordComparatorTest.java create mode 100644 ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/plan.json create mode 100644 ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/spec.json create mode 100644 ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/topology diff --git a/ksql-common/src/main/java/io/confluent/ksql/json/JsonMapper.java b/ksql-common/src/main/java/io/confluent/ksql/json/JsonMapper.java index 1a1418c4a726..625ef89bea7d 100644 --- a/ksql-common/src/main/java/io/confluent/ksql/json/JsonMapper.java +++ b/ksql-common/src/main/java/io/confluent/ksql/json/JsonMapper.java @@ -18,6 +18,7 @@ import com.fasterxml.jackson.core.JsonGenerator; import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; public enum JsonMapper { @@ -32,5 +33,6 @@ public enum JsonMapper { mapper.registerModule(new KsqlTypesSerializationModule()); mapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES); mapper.enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS); + mapper.setNodeFactory(JsonNodeFactory.withExactBigDecimals(true)); } } \ No newline at end of file diff --git a/ksql-common/src/test/java/io/confluent/ksql/json/JsonMapperTest.java b/ksql-common/src/test/java/io/confluent/ksql/json/JsonMapperTest.java index a1d9890f4dc4..f6d171931c65 100644 --- a/ksql-common/src/test/java/io/confluent/ksql/json/JsonMapperTest.java +++ b/ksql-common/src/test/java/io/confluent/ksql/json/JsonMapperTest.java @@ -15,12 +15,16 @@ package io.confluent.ksql.json; -import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.is; import com.fasterxml.jackson.core.JsonGenerator; import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.DecimalNode; +import java.math.BigDecimal; import org.junit.Test; public class JsonMapperTest { @@ -34,6 +38,35 @@ public void shouldNotAutoCloseTarget() { @Test public void shouldIgnoreUnknownProperties() { - assertThat(OBJECT_MAPPER.isEnabled(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES), is(false)); + assertThat(OBJECT_MAPPER.isEnabled(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES), + is(false)); + } + + @Test + public void shouldSerializeScientificDecimal() throws Exception { + // When: + final String json = OBJECT_MAPPER.writeValueAsString(new BigDecimal("1E+1")); + + // Then: + assertThat(json, is("1E+1")); + } + + @Test + public void shouldSerializeDecimalsWithoutLossOfTrailingZeros() throws Exception { + // When: + final String json = OBJECT_MAPPER.writeValueAsString(new BigDecimal("10.0")); + + // Then: + assertThat(json, is("10.0")); + } + + @Test + public void shouldDeserializeDecimalsWithoutLossOfTrailingZeros() throws Exception { + // When: + final JsonNode node = OBJECT_MAPPER.readTree("10.0"); + + // Then: + assertThat(node, is(instanceOf(DecimalNode.class))); + assertThat(node.decimalValue(), is(new BigDecimal("10.0"))); } } \ No newline at end of file diff --git a/ksql-functional-tests/src/main/java/io/confluent/ksql/test/serde/json/ValueSpecJsonSerdeSupplier.java b/ksql-functional-tests/src/main/java/io/confluent/ksql/test/serde/json/ValueSpecJsonSerdeSupplier.java index 9a8b18442045..e39968e37416 100644 --- a/ksql-functional-tests/src/main/java/io/confluent/ksql/test/serde/json/ValueSpecJsonSerdeSupplier.java +++ b/ksql-functional-tests/src/main/java/io/confluent/ksql/test/serde/json/ValueSpecJsonSerdeSupplier.java @@ -39,7 +39,9 @@ public class ValueSpecJsonSerdeSupplier implements SerdeSupplier { private static final ObjectMapper MAPPER = new ObjectMapper() - .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS); + .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS) + .setNodeFactory(JsonNodeFactory.withExactBigDecimals(true)); + private final boolean useSchemas; public ValueSpecJsonSerdeSupplier(final boolean useSchemas) { @@ -112,14 +114,17 @@ public Object deserialize(final String topicName, final byte[] data) { private static final class Converter { + private static final JsonNodeFactory JSON_NODE_FACTORY = JsonNodeFactory + .withExactBigDecimals(true); + private static final List> CONVERTORS = ImmutableList.of( - Converter.converter(Boolean.class, JsonNodeFactory.instance::booleanNode), - Converter.converter(Integer.class, JsonNodeFactory.instance::numberNode), - Converter.converter(Long.class, JsonNodeFactory.instance::numberNode), - Converter.converter(Float.class, JsonNodeFactory.instance::numberNode), - Converter.converter(Double.class, JsonNodeFactory.instance::numberNode), - Converter.converter(BigDecimal.class, JsonNodeFactory.instance::numberNode), - Converter.converter(String.class, JsonNodeFactory.instance::textNode), + Converter.converter(Boolean.class, JSON_NODE_FACTORY::booleanNode), + Converter.converter(Integer.class, JSON_NODE_FACTORY::numberNode), + Converter.converter(Long.class, JSON_NODE_FACTORY::numberNode), + Converter.converter(Float.class, JSON_NODE_FACTORY::numberNode), + Converter.converter(Double.class, JSON_NODE_FACTORY::numberNode), + Converter.converter(BigDecimal.class, JSON_NODE_FACTORY::numberNode), + Converter.converter(String.class, JSON_NODE_FACTORY::textNode), Converter.converter(Collection.class, Converter::handleCollection), Converter.converter(Map.class, Converter::handleMap) ); @@ -129,7 +134,7 @@ private static final class Converter { static JsonNode toJsonNode(final Object obj) { if (obj == null) { - return JsonNodeFactory.instance.nullNode(); + return JSON_NODE_FACTORY.nullNode(); } final List> candidates = CONVERTORS.stream() @@ -172,7 +177,7 @@ private JsonNode convert(final Object o) { } private static JsonNode handleCollection(final Collection collection) { - final ArrayNode list = JsonNodeFactory.instance.arrayNode(); + final ArrayNode list = JSON_NODE_FACTORY.arrayNode(); for (final Object element : collection) { list.add(toJsonNode(element)); } @@ -180,7 +185,7 @@ private static JsonNode handleCollection(final Collection collection) { } private static JsonNode handleMap(final Map map) { - final ObjectNode node = JsonNodeFactory.instance.objectNode(); + final ObjectNode node = JSON_NODE_FACTORY.objectNode(); if (map.isEmpty()) { return node; diff --git a/ksql-functional-tests/src/main/java/io/confluent/ksql/test/tools/ExpectedRecordComparator.java b/ksql-functional-tests/src/main/java/io/confluent/ksql/test/tools/ExpectedRecordComparator.java index 53a999b178d6..94e5230a94ec 100644 --- a/ksql-functional-tests/src/main/java/io/confluent/ksql/test/tools/ExpectedRecordComparator.java +++ b/ksql-functional-tests/src/main/java/io/confluent/ksql/test/tools/ExpectedRecordComparator.java @@ -23,7 +23,6 @@ import com.fasterxml.jackson.databind.node.TextNode; import com.google.common.collect.ImmutableMap; import java.math.BigDecimal; -import java.math.RoundingMode; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -107,33 +106,47 @@ private static boolean compareArray(final Object actualValue, final JsonNode exp private static boolean compareNumber(final Object actualValue, final JsonNode expectedValue) { final NumericNode expected = (NumericNode) expectedValue; + if (actualValue instanceof Integer) { return expected.intValue() == (Integer) actualValue; } + if (actualValue instanceof Long) { return expected.longValue() == (Long) actualValue; } + if (actualValue instanceof Double) { return expected.doubleValue() == (Double) actualValue; } + if (actualValue instanceof BigDecimal) { - if (!expected.isBigDecimal()) { - // we don't want to risk comparing a BigDecimal with something of - // lower precision - return false; - } + return compareDecimal((BigDecimal) actualValue, expected); + } + return false; + } - try { - return expected.decimalValue() - .setScale(((BigDecimal) actualValue).scale(), RoundingMode.UNNECESSARY) - .equals(actualValue); - } catch (final ArithmeticException e) { - // the scale of the expected value cannot match the scale of the actual value - // without rounding + private static boolean compareDecimal(final BigDecimal actualValue, final NumericNode expected) { + if (expected.isInt() || expected.isLong()) { + // Only match if actual has no decimal places: + if (actualValue.scale() > 0) { return false; } + + return expected.longValue() == actualValue.longValueExact(); + } + + if (!expected.isBigDecimal()) { + // we don't want to risk comparing a BigDecimal with something of lower precision + return false; + } + + try { + return expected.decimalValue().equals(actualValue); + } catch (final ArithmeticException e) { + // the scale of the expected value cannot match the scale of the actual value + // without rounding + return false; } - return false; } private static boolean compareText(final Object actualValue, final JsonNode expectedValue) { diff --git a/ksql-functional-tests/src/test/java/io/confluent/ksql/test/PlannedTestRewriterTest.java b/ksql-functional-tests/src/test/java/io/confluent/ksql/test/PlannedTestRewriterTest.java index 8f6be8f3042c..18b803a65655 100644 --- a/ksql-functional-tests/src/test/java/io/confluent/ksql/test/PlannedTestRewriterTest.java +++ b/ksql-functional-tests/src/test/java/io/confluent/ksql/test/PlannedTestRewriterTest.java @@ -20,6 +20,14 @@ import org.junit.Test; public class PlannedTestRewriterTest { + + /** + * Re-write ALL existing test plans. + * + *

You almost certainly do NOT want to do this as historical plans should be IMMUTABLE. + * The only time this is really valid is if you have fixed a bug in the testing framework + * and now need to correct bad historic test data. + */ @Test @Ignore public void rewritePlans() { diff --git a/ksql-functional-tests/src/test/java/io/confluent/ksql/test/serde/json/ValueSpecJsonSerdeSupplierTest.java b/ksql-functional-tests/src/test/java/io/confluent/ksql/test/serde/json/ValueSpecJsonSerdeSupplierTest.java new file mode 100644 index 000000000000..346849c739d5 --- /dev/null +++ b/ksql-functional-tests/src/test/java/io/confluent/ksql/test/serde/json/ValueSpecJsonSerdeSupplierTest.java @@ -0,0 +1,102 @@ +/* + * Copyright 2020 Confluent Inc. + * + * Licensed under the Confluent Community License (the "License"); you may not use + * this file except in compliance with the License. You may obtain a copy of the + * License at + * + * http://www.confluent.io/confluent-community-license + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package io.confluent.ksql.test.serde.json; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; +import io.confluent.ksql.json.JsonMapper; +import io.confluent.ksql.serde.json.JsonSerdeUtils; +import java.math.BigDecimal; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serializer; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class ValueSpecJsonSerdeSupplierTest { + + @Mock + private SchemaRegistryClient srClient; + + private ValueSpecJsonSerdeSupplier plainSerde; + private ValueSpecJsonSerdeSupplier srSerde; + + @Before + public void setUp() { + plainSerde = new ValueSpecJsonSerdeSupplier(false); + srSerde = new ValueSpecJsonSerdeSupplier(true); + } + + @Test + public void shouldSerializeDecimalsWithOutStrippingTrailingZeros_Plain() { + // Given: + final Serializer serializer = plainSerde.getSerializer(srClient); + + // When: + final byte[] bytes = serializer.serialize("t", new BigDecimal("10.0")); + + // Then: + assertThat(new String(bytes, UTF_8), is("10.0")); + } + + @Test + public void shouldDeserializeDecimalsWithoutStrippingTrailingZeros_Plain() { + // Given: + final Deserializer deserializer = plainSerde.getDeserializer(srClient); + + final byte[] bytes = "10.0".getBytes(UTF_8); + + // When: + final Object result = deserializer.deserialize("t", bytes); + + // Then: + assertThat(result, is(new BigDecimal("10.0"))); + } + + @Test + public void shouldSerializeDecimalsWithOutStrippingTrailingZeros_Sr() throws Exception { + // Given: + final Serializer serializer = srSerde.getSerializer(srClient); + + // When: + final byte[] bytes = serializer.serialize("t", new BigDecimal("10.0")); + + // Then: + assertThat(JsonSerdeUtils.readJsonSR(bytes, JsonMapper.INSTANCE.mapper, String.class), is("10.0")); + } + + @Test + public void shouldDeserializeDecimalsWithoutStrippingTrailingZeros_Sr() { + // Given: + final Deserializer deserializer = srSerde.getDeserializer(srClient); + + final byte[] jsonBytes = "10.0".getBytes(UTF_8); + final byte[] bytes = new byte[jsonBytes.length + JsonSerdeUtils.SIZE_OF_SR_PREFIX]; + System.arraycopy(jsonBytes, 0, bytes, JsonSerdeUtils.SIZE_OF_SR_PREFIX, jsonBytes.length); + + // When: + final Object result = deserializer.deserialize("t", bytes); + + // Then: + assertThat(result, is(new BigDecimal("10.0"))); + } +} \ No newline at end of file diff --git a/ksql-functional-tests/src/test/java/io/confluent/ksql/test/tools/ExpectedRecordComparatorTest.java b/ksql-functional-tests/src/test/java/io/confluent/ksql/test/tools/ExpectedRecordComparatorTest.java new file mode 100644 index 000000000000..15330e99868f --- /dev/null +++ b/ksql-functional-tests/src/test/java/io/confluent/ksql/test/tools/ExpectedRecordComparatorTest.java @@ -0,0 +1,102 @@ +/* + * Copyright 2020 Confluent Inc. + * + * Licensed under the Confluent Community License (the "License"); you may not use + * this file except in compliance with the License. You may obtain a copy of the + * License at + * + * http://www.confluent.io/confluent-community-license + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package io.confluent.ksql.test.tools; + +import static org.hamcrest.MatcherAssert.assertThat; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.DecimalNode; +import com.fasterxml.jackson.databind.node.DoubleNode; +import com.fasterxml.jackson.databind.node.IntNode; +import com.fasterxml.jackson.databind.node.LongNode; +import java.math.BigDecimal; +import java.math.RoundingMode; +import org.junit.Test; + +public class ExpectedRecordComparatorTest { + + @Test + public void shouldMatchActualBigDecimalWithZeroScaleToIntNode() { + assertMatch(new BigDecimal("10"), new IntNode(10)); + } + + @Test + public void shouldMatchActualBigDecimalWithNegativeScaleToIntNode() { + assertMatch(new BigDecimal("10").setScale(-1, RoundingMode.UNNECESSARY), new IntNode(10)); + } + + @Test + public void shouldNotMatchActualBigDecimalWithNonZeroScaleToIntNode() { + assertNoMatch(new BigDecimal("10.0"), new IntNode(10)); + } + + @Test + public void shouldMatchActualBigDecimalWithZeroScaleToLongNode() { + assertMatch(new BigDecimal("10"), new LongNode(10L)); + } + + @Test + public void shouldNotMatchActualBigDecimalWithNonZeroScaleToLongNode() { + assertNoMatch(new BigDecimal("10.0"), new LongNode(10)); + } + + @Test + public void shouldMatchActualBigDecimalWithNegativeScaleToLongNode() { + assertMatch(new BigDecimal("10").setScale(-1, RoundingMode.UNNECESSARY), new LongNode(10)); + } + + @Test + public void shouldNotMatchActualBigDecimalWithDoubleNode() { + assertNoMatch(new BigDecimal("10.1"), new DoubleNode(10.1)); + } + + @Test + public void shouldMatchActualBigDecimalWithMatchingDecimalNode() { + assertMatch(new BigDecimal("10.01"), new DecimalNode(new BigDecimal("10.01"))); + } + + @Test + public void shouldNotMatchActualBigDecimalWithDecimalNodeWithHigherScale() { + assertNoMatch(new BigDecimal("10.01"), new DecimalNode(new BigDecimal("10.010"))); + } + + @Test + public void shouldNotMatchActualBigDecimalWithDecimalNodeWithLowerScale() { + assertNoMatch(new BigDecimal("10.010"), new DecimalNode(new BigDecimal("10.01"))); + } + + private static void assertMatch(final Object actual, final JsonNode expected) { + assertThat( + "should match" + + System.lineSeparator() + + "Expected: " + expected + + System.lineSeparator() + + "Actual: " + actual, + ExpectedRecordComparator.matches(actual, expected) + ); + } + + private static void assertNoMatch(final Object actual, final JsonNode expected) { + assertThat( + "should not match" + + System.lineSeparator() + + "Expected: " + expected + + System.lineSeparator() + + "Actual: " + actual, + !ExpectedRecordComparator.matches(actual, expected) + ); + } +} \ No newline at end of file diff --git a/ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/plan.json b/ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/plan.json new file mode 100644 index 000000000000..50345d5ce707 --- /dev/null +++ b/ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/plan.json @@ -0,0 +1,147 @@ +{ + "plan" : [ { + "@type" : "ksqlPlanV1", + "statementText" : "CREATE STREAM INPUT (DEC DECIMAL(6, 4)) WITH (KAFKA_TOPIC='test', VALUE_FORMAT='JSON');", + "ddlCommand" : { + "@type" : "createStreamV1", + "sourceName" : "INPUT", + "schema" : "`ROWKEY` STRING KEY, `DEC` DECIMAL(6, 4)", + "keyField" : null, + "timestampColumn" : null, + "topicName" : "test", + "formats" : { + "keyFormat" : { + "format" : "KAFKA", + "properties" : { } + }, + "valueFormat" : { + "format" : "JSON", + "properties" : { } + }, + "options" : [ ] + }, + "windowInfo" : null + }, + "queryPlan" : null + }, { + "@type" : "ksqlPlanV1", + "statementText" : "CREATE STREAM OUTPUT AS SELECT *\nFROM INPUT INPUT\nEMIT CHANGES", + "ddlCommand" : { + "@type" : "createStreamV1", + "sourceName" : "OUTPUT", + "schema" : "`ROWKEY` STRING KEY, `DEC` DECIMAL(6, 4)", + "keyField" : null, + "timestampColumn" : null, + "topicName" : "OUTPUT", + "formats" : { + "keyFormat" : { + "format" : "KAFKA", + "properties" : { } + }, + "valueFormat" : { + "format" : "JSON", + "properties" : { } + }, + "options" : [ ] + }, + "windowInfo" : null + }, + "queryPlan" : { + "sources" : [ "INPUT" ], + "sink" : "OUTPUT", + "physicalPlan" : { + "@type" : "streamSinkV1", + "properties" : { + "queryContext" : "OUTPUT" + }, + "source" : { + "@type" : "streamSelectV1", + "properties" : { + "queryContext" : "Project" + }, + "source" : { + "@type" : "streamSourceV1", + "properties" : { + "queryContext" : "KsqlTopic/Source" + }, + "topicName" : "test", + "formats" : { + "keyFormat" : { + "format" : "KAFKA", + "properties" : { } + }, + "valueFormat" : { + "format" : "JSON", + "properties" : { } + }, + "options" : [ ] + }, + "timestampColumn" : null, + "sourceSchema" : "`ROWKEY` STRING KEY, `DEC` DECIMAL(6, 4)" + }, + "selectExpressions" : [ "DEC AS DEC" ] + }, + "formats" : { + "keyFormat" : { + "format" : "KAFKA", + "properties" : { } + }, + "valueFormat" : { + "format" : "JSON", + "properties" : { } + }, + "options" : [ ] + }, + "topicName" : "OUTPUT", + "timestampColumn" : null + }, + "queryId" : "CSAS_OUTPUT_0" + } + } ], + "configs" : { + "ksql.extension.dir" : "ext", + "ksql.streams.cache.max.bytes.buffering" : "0", + "ksql.security.extension.class" : null, + "ksql.transient.prefix" : "transient_", + "ksql.persistence.wrap.single.values" : "true", + "ksql.authorization.cache.expiry.time.secs" : "30", + "ksql.schema.registry.url" : "", + "ksql.streams.default.deserialization.exception.handler" : "io.confluent.ksql.errors.LogMetricAndContinueExceptionHandler", + "ksql.output.topic.name.prefix" : "", + "ksql.streams.auto.offset.reset" : "earliest", + "ksql.query.pull.enable.standby.reads" : "false", + "ksql.connect.url" : "http://localhost:8083", + "ksql.service.id" : "some.ksql.service.id", + "ksql.internal.topic.min.insync.replicas" : "1", + "ksql.streams.shutdown.timeout.ms" : "300000", + "ksql.new.api.enabled" : "false", + "ksql.streams.state.dir" : "/var/folders/2d/3pt97ylj3zngd51bwl91bl3r0000gp/T/confluent6088080813505006712", + "ksql.internal.topic.replicas" : "1", + "ksql.insert.into.values.enabled" : "true", + "ksql.query.pull.max.allowed.offset.lag" : "9223372036854775807", + "ksql.streams.default.production.exception.handler" : "io.confluent.ksql.errors.ProductionExceptionHandlerUtil$LogAndFailProductionExceptionHandler", + "ksql.access.validator.enable" : "auto", + "ksql.streams.bootstrap.servers" : "localhost:0", + "ksql.streams.commit.interval.ms" : "2000", + "ksql.metric.reporters" : "", + "ksql.streams.auto.commit.interval.ms" : "0", + "ksql.metrics.extension" : null, + "ksql.streams.topology.optimization" : "all", + "ksql.query.pull.streamsstore.rebalancing.timeout.ms" : "10000", + "ksql.hidden.topics" : "_confluent.*,__confluent.*,_schemas,__consumer_offsets,__transaction_state,connect-configs,connect-offsets,connect-status,connect-statuses", + "ksql.streams.num.stream.threads" : "4", + "ksql.timestamp.throw.on.invalid" : "false", + "ksql.authorization.cache.max.entries" : "10000", + "ksql.metrics.tags.custom" : "", + "ksql.pull.queries.enable" : "true", + "ksql.udfs.enabled" : "true", + "ksql.udf.enable.security.manager" : "true", + "ksql.connect.worker.config" : "", + "ksql.any.key.name.enabled" : "false", + "ksql.sink.window.change.log.additional.retention" : "1000000", + "ksql.readonly.topics" : "_confluent.*,__confluent.*,_schemas,__consumer_offsets,__transaction_state,connect-configs,connect-offsets,connect-status,connect-statuses", + "ksql.udf.collect.metrics" : "false", + "ksql.persistent.prefix" : "query_", + "ksql.query.persistent.active.limit" : "2147483647" + } +} \ No newline at end of file diff --git a/ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/spec.json b/ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/spec.json new file mode 100644 index 000000000000..6951f5e8f45f --- /dev/null +++ b/ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/spec.json @@ -0,0 +1,82 @@ +{ + "version" : "6.0.0", + "timestamp" : 1583419431528, + "schemas" : { + "CSAS_OUTPUT_0.KsqlTopic.Source" : "STRUCT NOT NULL", + "CSAS_OUTPUT_0.OUTPUT" : "STRUCT NOT NULL" + }, + "inputs" : [ { + "topic" : "test", + "key" : "", + "value" : { + "DEC" : 10 + } + }, { + "topic" : "test", + "key" : "", + "value" : { + "DEC" : 1 + } + }, { + "topic" : "test", + "key" : "", + "value" : { + "DEC" : 0.1 + } + }, { + "topic" : "test", + "key" : "", + "value" : { + "DEC" : 0.01 + } + }, { + "topic" : "test", + "key" : "", + "value" : { + "DEC" : 0.001 + } + }, { + "topic" : "test", + "key" : "", + "value" : { + "DEC" : 0.0001 + } + } ], + "outputs" : [ { + "topic" : "OUTPUT", + "key" : "", + "value" : { + "DEC" : 10 + } + }, { + "topic" : "OUTPUT", + "key" : "", + "value" : { + "DEC" : 1 + } + }, { + "topic" : "OUTPUT", + "key" : "", + "value" : { + "DEC" : 0.1 + } + }, { + "topic" : "OUTPUT", + "key" : "", + "value" : { + "DEC" : 0.01 + } + }, { + "topic" : "OUTPUT", + "key" : "", + "value" : { + "DEC" : 0.001 + } + }, { + "topic" : "OUTPUT", + "key" : "", + "value" : { + "DEC" : 0.0001 + } + } ] +} \ No newline at end of file diff --git a/ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/topology b/ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/topology new file mode 100644 index 000000000000..9396b997ae94 --- /dev/null +++ b/ksql-functional-tests/src/test/resources/historical_plans/decimal_-_JSON_scale_in_data_less_than_scale_in_type/6.0.0_1583419431528/topology @@ -0,0 +1,13 @@ +Topologies: + Sub-topology: 0 + Source: KSTREAM-SOURCE-0000000000 (topics: [test]) + --> KSTREAM-TRANSFORMVALUES-0000000001 + Processor: KSTREAM-TRANSFORMVALUES-0000000001 (stores: []) + --> Project + <-- KSTREAM-SOURCE-0000000000 + Processor: Project (stores: []) + --> KSTREAM-SINK-0000000003 + <-- KSTREAM-TRANSFORMVALUES-0000000001 + Sink: KSTREAM-SINK-0000000003 (topic: OUTPUT) + <-- Project + diff --git a/ksql-functional-tests/src/test/resources/query-validation-tests/decimal.json b/ksql-functional-tests/src/test/resources/query-validation-tests/decimal.json index ba0fe65f25b1..6b17c2f77301 100644 --- a/ksql-functional-tests/src/test/resources/query-validation-tests/decimal.json +++ b/ksql-functional-tests/src/test/resources/query-validation-tests/decimal.json @@ -8,10 +8,10 @@ "CREATE STREAM TEST2 AS SELECT * FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": "10.1234512345123451234"} + {"topic": "test", "value": "10.1234512345123451234"} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": "10.1234512345123451234"} + {"topic": "TEST2", "value": "10.1234512345123451234"} ] }, { @@ -21,10 +21,10 @@ "CREATE STREAM TEST2 AS SELECT * FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"DEC": "10.1234512345123451234"}} + {"topic": "test", "value": {"DEC": "10.1234512345123451234"}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"DEC": "10.1234512345123451234"}} + {"topic": "TEST2", "value": {"DEC": "10.1234512345123451234"}} ] }, { @@ -38,12 +38,64 @@ "CREATE STREAM TEST2 AS SELECT * FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"DEC": "10.1234512345123451234"}} + {"topic": "test", "value": {"DEC": "10.1234512345123451234"}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"DEC": "10.1234512345123451234"}} + {"topic": "TEST2", "value": {"DEC": "10.1234512345123451234"}} ] }, + { + "name": "JSON scale in data less than scale in type", + "statements": [ + "CREATE STREAM INPUT (dec DECIMAL(6,4)) WITH (kafka_topic='test', value_format='JSON');", + "CREATE STREAM OUTPUT AS SELECT * FROM INPUT;" + ], + "inputs": [ + {"topic": "test", "value": {"DEC": 10}}, + {"topic": "test", "value": {"DEC": 1}}, + {"topic": "test", "value": {"DEC": 0.1}}, + {"topic": "test", "value": {"DEC": 0.01}}, + {"topic": "test", "value": {"DEC": 0.001}}, + {"topic": "test", "value": {"DEC": 0.0001}} + ], + "outputs": [ + {"topic": "OUTPUT", "value": {"DEC": 10}}, + {"topic": "OUTPUT", "value": {"DEC": 1}}, + {"topic": "OUTPUT", "value": {"DEC": 0.1}}, + {"topic": "OUTPUT", "value": {"DEC": 0.01}}, + {"topic": "OUTPUT", "value": {"DEC": 0.001}}, + {"topic": "OUTPUT", "value": {"DEC": 0.0001}} + ], + "post": { + "sources": [ + {"name": "INPUT", "type": "stream", "schema": "ROWKEY STRING KEY, DEC DECIMAL(6,4)"}, + {"name": "OUTPUT", "type": "stream", "schema": "ROWKEY STRING KEY, DEC DECIMAL(6,4)"} + ] + } + }, + { + "enabled": false, + "name": "JSON should not trim trailing zeros", + "comment": "Disabled until https://github.com/confluentinc/ksql/issues/4710 is done", + "statements": [ + "CREATE STREAM INPUT (dec DECIMAL(6,4)) WITH (kafka_topic='test', value_format='JSON');", + "CREATE STREAM OUTPUT AS SELECT * FROM INPUT;" + ], + "inputs": [ + {"topic": "test", "value": {"DEC": 10.0}}, + {"topic": "test", "value": {"DEC": 1.0000}} + ], + "outputs": [ + {"topic": "OUTPUT", "value": {"DEC": 10.0}}, + {"topic": "test", "value": {"DEC": 1.0000}} + ], + "post": { + "sources": [ + {"name": "INPUT", "type": "stream", "schema": "ROWKEY STRING KEY, DEC DECIMAL(6,4)"}, + {"name": "OUTPUT", "type": "stream", "schema": "ROWKEY STRING KEY, DEC DECIMAL(6,4)"} + ] + } + }, { "name": "negation", "statements": [ @@ -51,10 +103,10 @@ "CREATE STREAM TEST2 AS SELECT -dec AS negated FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"DEC": "10.12345"}} + {"topic": "test", "value": {"DEC": "10.12345"}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"NEGATED": "-10.12345"}} + {"topic": "TEST2", "value": {"NEGATED": "-10.12345"}} ] }, { @@ -64,14 +116,14 @@ "CREATE STREAM TEST2 AS SELECT (a + b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "5.10"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "-5.00"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "0.00"}} + {"topic": "test", "value": {"A": "10.01", "B": "5.10"}}, + {"topic": "test", "value": {"A": "10.01", "B": "-5.00"}}, + {"topic": "test", "value": {"A": "10.01", "B": "0.00"}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": "15.11"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": "5.01"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": "10.01"}} + {"topic": "TEST2", "value": {"RESULT": "15.11"}}, + {"topic": "TEST2", "value": {"RESULT": "5.01"}}, + {"topic": "TEST2", "value": {"RESULT": "10.01"}} ] }, { @@ -81,14 +133,14 @@ "CREATE STREAM TEST2 AS SELECT (a + b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": 5.1}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": -5.0}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": 0.0}} + {"topic": "test", "value": {"A": "10.01", "B": 5.1}}, + {"topic": "test", "value": {"A": "10.01", "B": -5.0}}, + {"topic": "test", "value": {"A": "10.01", "B": 0.0}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": 15.11}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": 5.01}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": 10.01}} + {"topic": "TEST2", "value": {"RESULT": 15.11}}, + {"topic": "TEST2", "value": {"RESULT": 5.01}}, + {"topic": "TEST2", "value": {"RESULT": 10.01}} ] }, { @@ -98,14 +150,14 @@ "CREATE STREAM TEST2 AS SELECT (a + b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": 5}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": -5}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": 0}} + {"topic": "test", "value": {"A": "10.01", "B": 5}}, + {"topic": "test", "value": {"A": "10.01", "B": -5}}, + {"topic": "test", "value": {"A": "10.01", "B": 0}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": "15.01"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": "5.01"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": "10.01"}} + {"topic": "TEST2", "value": {"RESULT": "15.01"}}, + {"topic": "TEST2", "value": {"RESULT": "5.01"}}, + {"topic": "TEST2", "value": {"RESULT": "10.01"}} ] }, { @@ -115,10 +167,10 @@ "CREATE STREAM TEST2 AS SELECT (a + a + b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "5.10"}} + {"topic": "test", "value": {"A": "10.01", "B": "5.10"}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": "25.12"}} + {"topic": "TEST2", "value": {"RESULT": "25.12"}} ] }, { @@ -128,14 +180,14 @@ "CREATE STREAM TEST2 AS SELECT (a - b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "5.10"}}, - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "-5.00"}}, - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "0.00"}} + {"topic": "test", "value": {"A": "10.10", "B": "5.10"}}, + {"topic": "test", "value": {"A": "10.10", "B": "-5.00"}}, + {"topic": "test", "value": {"A": "10.10", "B": "0.00"}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": "05.00"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": "15.10"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": "10.10"}} + {"topic": "TEST2", "value": {"RESULT": "05.00"}}, + {"topic": "TEST2", "value": {"RESULT": "15.10"}}, + {"topic": "TEST2", "value": {"RESULT": "10.10"}} ] }, { @@ -145,14 +197,14 @@ "CREATE STREAM TEST2 AS SELECT (a * b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "02.00"}}, - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "-02.00"}}, - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "00.00"}} + {"topic": "test", "value": {"A": "10.10", "B": "02.00"}}, + {"topic": "test", "value": {"A": "10.10", "B": "-02.00"}}, + {"topic": "test", "value": {"A": "10.10", "B": "00.00"}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": "20.2000"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": "-20.2000"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": "00.0000"}} + {"topic": "TEST2", "value": {"RESULT": "20.2000"}}, + {"topic": "TEST2", "value": {"RESULT": "-20.2000"}}, + {"topic": "TEST2", "value": {"RESULT": "00.0000"}} ] }, { @@ -165,14 +217,14 @@ "The last record causes division by zero, the error is logged and a null value is output" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "02.00"}}, - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "-02.00"}}, - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "00.00"}} + {"topic": "test", "value": {"A": "10.10", "B": "02.00"}}, + {"topic": "test", "value": {"A": "10.10", "B": "-02.00"}}, + {"topic": "test", "value": {"A": "10.10", "B": "00.00"}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": "005.0500000"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": "-005.0500000"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": null}} + {"topic": "TEST2", "value": {"RESULT": "005.0500000"}}, + {"topic": "TEST2", "value": {"RESULT": "-005.0500000"}}, + {"topic": "TEST2", "value": {"RESULT": null}} ] }, { @@ -185,14 +237,14 @@ "The last record causes modulo by zero, the error is logged and a null value is output" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "02.00"}}, - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "-02.00"}}, - {"topic": "test", "key": "0", "value": {"A": "10.10", "B": "00.00"}} + {"topic": "test", "value": {"A": "10.10", "B": "02.00"}}, + {"topic": "test", "value": {"A": "10.10", "B": "-02.00"}}, + {"topic": "test", "value": {"A": "10.10", "B": "00.00"}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": "0.10"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": "0.10"}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": null}} + {"topic": "TEST2", "value": {"RESULT": "0.10"}}, + {"topic": "TEST2", "value": {"RESULT": "0.10"}}, + {"topic": "TEST2", "value": {"RESULT": null}} ] }, { @@ -202,16 +254,16 @@ "CREATE STREAM TEST2 AS SELECT (a = b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "12.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": null}} + {"topic": "test", "value": {"A": "10.01", "B": "10.01"}}, + {"topic": "test", "value": {"A": "10.01", "B": "12.01"}}, + {"topic": "test", "value": {"A": null, "B": "10.01"}}, + {"topic": "test", "value": {"A": null, "B": null}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}} + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}} ] }, { @@ -221,16 +273,16 @@ "CREATE STREAM TEST2 AS SELECT (a <> b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "12.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": null}} + {"topic": "test", "value": {"A": "10.01", "B": "10.01"}}, + {"topic": "test", "value": {"A": "10.01", "B": "12.01"}}, + {"topic": "test", "value": {"A": null, "B": "10.01"}}, + {"topic": "test", "value": {"A": null, "B": null}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}} + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}} ] }, { @@ -240,16 +292,16 @@ "CREATE STREAM TEST2 AS SELECT (a IS DISTINCT FROM b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "12.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": null}} + {"topic": "test", "value": {"A": "10.01", "B": "10.01"}}, + {"topic": "test", "value": {"A": "10.01", "B": "12.01"}}, + {"topic": "test", "value": {"A": null, "B": "10.01"}}, + {"topic": "test", "value": {"A": null, "B": null}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}} + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": false}} ] }, { @@ -259,16 +311,16 @@ "CREATE STREAM TEST2 AS SELECT (a < b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "12.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": null}} + {"topic": "test", "value": {"A": "10.01", "B": "10.01"}}, + {"topic": "test", "value": {"A": "10.01", "B": "12.01"}}, + {"topic": "test", "value": {"A": null, "B": "10.01"}}, + {"topic": "test", "value": {"A": null, "B": null}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}} + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}} ] }, { @@ -278,16 +330,16 @@ "CREATE STREAM TEST2 AS SELECT (a < b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "10.010"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "10.012"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": "10.010"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": null}} + {"topic": "test", "value": {"A": "10.01", "B": "10.010"}}, + {"topic": "test", "value": {"A": "10.01", "B": "10.012"}}, + {"topic": "test", "value": {"A": null, "B": "10.010"}}, + {"topic": "test", "value": {"A": null, "B": null}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}} + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}} ] }, { @@ -297,18 +349,18 @@ "CREATE STREAM TEST2 AS SELECT (a <= b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "03.01"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "12.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": null}} + {"topic": "test", "value": {"A": "10.01", "B": "03.01"}}, + {"topic": "test", "value": {"A": "10.01", "B": "10.01"}}, + {"topic": "test", "value": {"A": "10.01", "B": "12.01"}}, + {"topic": "test", "value": {"A": null, "B": "10.01"}}, + {"topic": "test", "value": {"A": null, "B": null}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}} + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}} ] }, { @@ -318,18 +370,18 @@ "CREATE STREAM TEST2 AS SELECT (a >= b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "03.01"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "12.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": null}} + {"topic": "test", "value": {"A": "10.01", "B": "03.01"}}, + {"topic": "test", "value": {"A": "10.01", "B": "10.01"}}, + {"topic": "test", "value": {"A": "10.01", "B": "12.01"}}, + {"topic": "test", "value": {"A": null, "B": "10.01"}}, + {"topic": "test", "value": {"A": null, "B": null}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}} + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}} ] }, { @@ -339,18 +391,18 @@ "CREATE STREAM TEST2 AS SELECT (a > b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "03.01"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": "12.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": "10.01"}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": null}} + {"topic": "test", "value": {"A": "10.01", "B": "03.01"}}, + {"topic": "test", "value": {"A": "10.01", "B": "10.01"}}, + {"topic": "test", "value": {"A": "10.01", "B": "12.01"}}, + {"topic": "test", "value": {"A": null, "B": "10.01"}}, + {"topic": "test", "value": {"A": null, "B": null}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}} + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}} ] }, { @@ -360,16 +412,16 @@ "CREATE STREAM TEST2 AS SELECT (a < b) AS RESULT FROM TEST;" ], "inputs": [ - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": 1}}, - {"topic": "test", "key": "0", "value": {"A": "10.01", "B": 12}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": 12}}, - {"topic": "test", "key": "0", "value": {"A": null, "B": null}} + {"topic": "test", "value": {"A": "10.01", "B": 1}}, + {"topic": "test", "value": {"A": "10.01", "B": 12}}, + {"topic": "test", "value": {"A": null, "B": 12}}, + {"topic": "test", "value": {"A": null, "B": null}} ], "outputs": [ - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": true}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}}, - {"topic": "TEST2", "key": "0", "value": {"RESULT": false}} + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": true}}, + {"topic": "TEST2", "value": {"RESULT": false}}, + {"topic": "TEST2", "value": {"RESULT": false}} ] } ] diff --git a/ksql-functional-tests/src/test/resources/query-validation-tests/sum.json b/ksql-functional-tests/src/test/resources/query-validation-tests/sum.json index ff8598c5d7ad..9206006e9278 100644 --- a/ksql-functional-tests/src/test/resources/query-validation-tests/sum.json +++ b/ksql-functional-tests/src/test/resources/query-validation-tests/sum.json @@ -207,8 +207,8 @@ "outputs": [ {"topic": "OUTPUT", "key": "0", "value": {"SUM_VAL":2.0}}, {"topic": "OUTPUT", "key": "0", "value": {"SUM_VAL":6.0}}, - {"topic": "OUTPUT", "key": "0", "value": {"SUM_VAL":922337203692.0}}, - {"topic": "OUTPUT", "key": "0", "value": {"SUM_VAL":922337203694.0}} + {"topic": "OUTPUT", "key": "0", "value": {"SUM_VAL":922337203692}}, + {"topic": "OUTPUT", "key": "0", "value": {"SUM_VAL":922337203694}} ] }, { diff --git a/ksql-serde/src/main/java/io/confluent/ksql/serde/json/JsonSerdeUtils.java b/ksql-serde/src/main/java/io/confluent/ksql/serde/json/JsonSerdeUtils.java index 07ffb51c05cc..dd366353fa91 100644 --- a/ksql-serde/src/main/java/io/confluent/ksql/serde/json/JsonSerdeUtils.java +++ b/ksql-serde/src/main/java/io/confluent/ksql/serde/json/JsonSerdeUtils.java @@ -35,7 +35,8 @@ public final class JsonSerdeUtils { // the JsonSchemaConverter adds a magic NULL byte and 4 bytes for the // schema ID at the start of the message - private static final int SIZE_OF_SR_PREFIX = Byte.BYTES + Integer.BYTES; + public static final int SIZE_OF_SR_PREFIX = Byte.BYTES + Integer.BYTES; + public static final int MAGIC_BYTE = 0x00; private JsonSerdeUtils() { } @@ -79,7 +80,7 @@ static boolean hasMagicByte(@Nonnull final byte[] json) { // (https://tools.ietf.org/html/rfc7159#section-2) valid JSON should not // start with 0x00 - the only "insignificant" characters allowed are // 0x20, 0x09, 0x0A and 0x0D - return json.length > 0 && json[0] == 0x00; + return json.length > 0 && json[0] == MAGIC_BYTE; } static PersistenceSchema validateSchema(final PersistenceSchema schema) { diff --git a/ksql-serde/src/main/java/io/confluent/ksql/serde/json/KsqlJsonDeserializer.java b/ksql-serde/src/main/java/io/confluent/ksql/serde/json/KsqlJsonDeserializer.java index 05f75fd1c6e7..dfd60095fa1f 100644 --- a/ksql-serde/src/main/java/io/confluent/ksql/serde/json/KsqlJsonDeserializer.java +++ b/ksql-serde/src/main/java/io/confluent/ksql/serde/json/KsqlJsonDeserializer.java @@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; import com.fasterxml.jackson.databind.node.NullNode; import com.fasterxml.jackson.databind.node.NumericNode; import com.fasterxml.jackson.databind.node.ObjectNode; @@ -56,7 +57,8 @@ public class KsqlJsonDeserializer implements Deserializer { private static final Logger LOG = LoggerFactory.getLogger(KsqlJsonDeserializer.class); private static final SqlSchemaFormatter FORMATTER = new SqlSchemaFormatter(word -> false); private static final ObjectMapper MAPPER = new ObjectMapper() - .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS); + .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS) + .setNodeFactory(JsonNodeFactory.withExactBigDecimals(true)); private static final Schema STRING_ARRAY = SchemaBuilder .array(Schema.OPTIONAL_STRING_SCHEMA).build(); diff --git a/ksql-serde/src/test/java/io/confluent/ksql/serde/json/KsqlJsonDeserializerTest.java b/ksql-serde/src/test/java/io/confluent/ksql/serde/json/KsqlJsonDeserializerTest.java index 5a77d5e09ea6..e0776a56886c 100644 --- a/ksql-serde/src/test/java/io/confluent/ksql/serde/json/KsqlJsonDeserializerTest.java +++ b/ksql-serde/src/test/java/io/confluent/ksql/serde/json/KsqlJsonDeserializerTest.java @@ -15,6 +15,7 @@ package io.confluent.ksql.serde.json; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.endsWith; @@ -26,9 +27,11 @@ import static org.junit.internal.matchers.ThrowableMessageMatcher.hasMessage; import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.BooleanNode; import com.fasterxml.jackson.databind.node.IntNode; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import io.confluent.ksql.schema.ksql.PersistenceSchema; @@ -97,17 +100,22 @@ public class KsqlJsonDeserializerTest { .put("caseField", 1L) .build(); - private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper() + .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS) + .setNodeFactory(JsonNodeFactory.withExactBigDecimals(true)); @Rule public final ExpectedException expectedException = ExpectedException.none(); - @Parameters + @Parameters(name = "{0}") public static Collection data() { - return Arrays.asList(new Object[][]{{false}, {true}}); + return Arrays.asList(new Object[][]{{"Plain JSON", false}, {"Magic byte prefixed", true}}); } @Parameter + public String suiteName; + + @Parameter(1) public boolean useSchemas; private Struct expectedOrder; @@ -457,7 +465,7 @@ public void shouldDeserializedJsonText() { final Map validCoercions = ImmutableMap.builder() .put("true", "true") .put("42", "42") - .put("42.000", "42") + .put("42.000", "42.000") .put("42.001", "42.001") .put("\"just a string\"", "just a string") .put("{\"json\": \"object\"}", "{\"json\":\"object\"}") @@ -498,6 +506,34 @@ public void shouldDeserializedJsonNumberAsBigDecimal() { }); } + @Test + public void shouldDeserializeDecimalsWithoutStrippingTrailingZeros() { + // Given: + givenDeserializerForSchema(DecimalUtil.builder(3,1).build()); + + final byte[] bytes = addMagic("10.0".getBytes(UTF_8)); + + // When: + final Object result = deserializer.deserialize(SOME_TOPIC, bytes); + + // Then: + assertThat(result, is(new BigDecimal("10.0"))); + } + + @Test + public void shouldDeserializeScientificNotation() { + // Given: + givenDeserializerForSchema(DecimalUtil.builder(3,1).build()); + + final byte[] bytes = addMagic("1E+1".getBytes(UTF_8)); + + // When: + final Object result = deserializer.deserialize(SOME_TOPIC, bytes); + + // Then: + assertThat(result, is(new BigDecimal("1e+1"))); + } + @Test public void shouldThrowIfCanNotCoerceToBigDecimal() { // Given: From fc3e93668d154f5900ba1ae5f49124a2317f4a05 Mon Sep 17 00:00:00 2001 From: Andy Coates Date: Thu, 5 Mar 2020 18:02:18 +0000 Subject: [PATCH 2/5] chore: fix test --- .../src/test/resources/query-validation-tests/decimal.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ksql-functional-tests/src/test/resources/query-validation-tests/decimal.json b/ksql-functional-tests/src/test/resources/query-validation-tests/decimal.json index 6b17c2f77301..dd741d4dd9fc 100644 --- a/ksql-functional-tests/src/test/resources/query-validation-tests/decimal.json +++ b/ksql-functional-tests/src/test/resources/query-validation-tests/decimal.json @@ -87,7 +87,7 @@ ], "outputs": [ {"topic": "OUTPUT", "value": {"DEC": 10.0}}, - {"topic": "test", "value": {"DEC": 1.0000}} + {"topic": "OUTPUT", "value": {"DEC": 1.0000}} ], "post": { "sources": [ From 9fc024047d3b99116e1b28d5e0aa6f92133f83df Mon Sep 17 00:00:00 2001 From: Andy Coates Date: Thu, 5 Mar 2020 18:20:50 +0000 Subject: [PATCH 3/5] chore: tests add some tests to remind us to finish this work off once Connect fixed --- .../serde/json/KsqlJsonSerializerTest.java | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/ksql-serde/src/test/java/io/confluent/ksql/serde/json/KsqlJsonSerializerTest.java b/ksql-serde/src/test/java/io/confluent/ksql/serde/json/KsqlJsonSerializerTest.java index 93716d5ad48f..0dba7dcb11c7 100644 --- a/ksql-serde/src/test/java/io/confluent/ksql/serde/json/KsqlJsonSerializerTest.java +++ b/ksql-serde/src/test/java/io/confluent/ksql/serde/json/KsqlJsonSerializerTest.java @@ -52,6 +52,7 @@ import org.apache.kafka.connect.data.Struct; import org.hamcrest.CoreMatchers; import org.junit.Before; +import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; @@ -327,6 +328,36 @@ public void shouldSerializeDecimal() { assertThat(asJsonString(bytes), is("1.234567890123456789")); } + @Ignore // Until https://github.com/confluentinc/ksql/issues/4710 is done. + @Test + public void shouldSerializeDecimalsWithoutStrippingTrailingZeros() { + // Given: + givenSerializerForSchema(DecimalUtil.builder(3,1).build()); + + // When: + final byte[] bytes = serializer.serialize(SOME_TOPIC, new BigDecimal("12.0")); + + // Then: + assertThat(asJsonString(bytes), is("12.0")); + } + + /* + When this test starts failing it's because https://issues.apache.org/jira/browse/KAFKA-9667 + has been fixed. This test should be removed. The above test, and the matching test in decimal.json + should also be enabled. (Search for https://github.com/confluentinc/ksql/issues/4710 in the code). + */ + @Test + public void currentlyDoesStripTrailingZerosButShouldNot() { + // Given: + givenSerializerForSchema(DecimalUtil.builder(3,1).build()); + + // When: + final byte[] bytes = serializer.serialize(SOME_TOPIC, new BigDecimal("12.0")); + + // Then: + assertThat(asJsonString(bytes), is("12")); + } + @Test public void shouldThrowIfNotDouble() { // Given: From b8d6b89f426e540ed079a538218b0066a6b518de Mon Sep 17 00:00:00 2001 From: Andy Coates Date: Fri, 6 Mar 2020 10:25:55 +0000 Subject: [PATCH 4/5] chore: tests Add client test to ensure client doesn't strip those trailing zeros. --- .../ksql/rest/client/KsqlClientTest.java | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/ksql-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlClientTest.java b/ksql-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlClientTest.java index 28baec0af3fc..ed5db3eee979 100644 --- a/ksql-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlClientTest.java +++ b/ksql-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlClientTest.java @@ -50,6 +50,7 @@ import io.vertx.core.http.HttpMethod; import io.vertx.core.http.HttpServerOptions; import io.vertx.core.net.JksOptions; +import java.math.BigDecimal; import java.net.URI; import java.nio.charset.StandardCharsets; import java.util.ArrayList; @@ -313,6 +314,26 @@ public void shouldPostQueryRequest() { assertThat(response.get(), is(expectedResponse)); } + @Test + public void shouldNotTrimTrailingZerosOnDecimalDeserialization() { + // Given: + server.setResponseBuffer(Buffer.buffer("" + + "[" + + "{\"row\": {\"columns\": [1.000, 12.100]}}" + + "]" + )); + + // When: + final KsqlTarget target = ksqlClient.target(serverUri); + RestResponse> response = target.postQueryRequest( + "some sql", Collections.emptyMap(), Optional.of(321L)); + + // Then: + assertThat(response.get(), is(ImmutableList.of( + StreamedRow.row(GenericRow.genericRow(new BigDecimal("1.000"), new BigDecimal("12.100"))) + ))); + } + @Test public void shouldPostQueryRequestStreamed() throws Exception { From d95eaefb762e55785f97ec9037f5ab342e5bbc5d Mon Sep 17 00:00:00 2001 From: Andy Coates Date: Fri, 6 Mar 2020 15:09:22 +0000 Subject: [PATCH 5/5] chore: correct old invalid test data --- .../5.5.0_1581572103953/spec.json | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ksql-functional-tests/src/test/resources/historical_plans/sum_-_sum_list_of_doubles_into_a_single_double/5.5.0_1581572103953/spec.json b/ksql-functional-tests/src/test/resources/historical_plans/sum_-_sum_list_of_doubles_into_a_single_double/5.5.0_1581572103953/spec.json index f80934bc2029..66369c89fc5e 100644 --- a/ksql-functional-tests/src/test/resources/historical_plans/sum_-_sum_list_of_doubles_into_a_single_double/5.5.0_1581572103953/spec.json +++ b/ksql-functional-tests/src/test/resources/historical_plans/sum_-_sum_list_of_doubles_into_a_single_double/5.5.0_1581572103953/spec.json @@ -48,13 +48,13 @@ "topic" : "OUTPUT", "key" : "0", "value" : { - "SUM_VAL" : 922337203692.0 + "SUM_VAL" : 922337203692 } }, { "topic" : "OUTPUT", "key" : "0", "value" : { - "SUM_VAL" : 922337203694.0 + "SUM_VAL" : 922337203694 } } ] } \ No newline at end of file