Skip to content
New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Jump to bottom

Fix: Consistency between inserted rowkey and key values when inserting into a table with a key #3381

Merged
merged 6 commits into from Sep 18, 2019
Merged

Fix: Consistency between inserted rowkey and key values when inserting into a table with a key #3381

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d49f282
fix:#3347 If the table has an key and a value for it has been provide…
Sep 18, 2019
f141cc1
fix: When ROWKEY and ID are both specified in insert into, then check…
Sep 18, 2019
fb05124
fix: Add a bunch of RQTT tests to test combinations of inserts with i…
Sep 18, 2019
5608933
fix:nit
Sep 18, 2019
8535847
fix:nit
Sep 18, 2019
6a6a05e
fix:nits
Sep 18, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
17 changes: 10 additions & 7 deletions ksql-engine/src/main/java/io/confluent/ksql/engine/InsertValuesExecutor.java
View file
Open in desktop
Expand Up @@ -323,13 +323,16 @@ private static void handleExplicitKeyField(
final Object rowKeyValue = values.get(SchemaUtil.ROWKEY_NAME);

if (keyValue != null ^ rowKeyValue != null) {
values.putIfAbsent(key, rowKeyValue);
values.putIfAbsent(SchemaUtil.ROWKEY_NAME, keyValue);
} else if (!Objects.equals(keyValue, rowKeyValue)) {
throw new KsqlException(
String.format(
"Expected ROWKEY and %s to match but got %s and %s respectively.",
key, rowKeyValue, keyValue));
if (keyValue == null) {
values.putIfAbsent(key, rowKeyValue);
purplefox marked this conversation as resolved.
Show resolved Hide resolved
} else {
// Note, ROWKEY must always be a String, but key might not be so we need toString()
purplefox marked this conversation as resolved.
Show resolved Hide resolved
values.putIfAbsent(SchemaUtil.ROWKEY_NAME, keyValue.toString());
}
} else if (keyValue != null && !Objects.equals(keyValue.toString(), rowKeyValue)) {
throw new KsqlException(String.format(
"Expected ROWKEY and %s to match but got %s and %s respectively.",
key, rowKeyValue, keyValue));
}
}
}
Expand Down
144 changes: 144 additions & 0 deletions ksql-functional-tests/src/test/resources/rest-query-validation-tests/insert-values.json
View file
Open in desktop
Expand Up @@ -66,6 +66,138 @@
{"topic": "test_topic", "key": "10", "value": {"ID": null}}
]
},
{
"name": "rowkey should be set when stream has int key and only id specified in insert",
purplefox marked this conversation as resolved.
Show resolved Hide resolved
"statements": [
"CREATE STREAM TEST (ID INT) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ID) VALUES (10);"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "10", "value": {"ID": 10}}
]
},
{
"name": "rowkey should be set when stream has String key and only id specified in insert",
"statements": [
"CREATE STREAM TEST (ID VARCHAR) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ID) VALUES ('10');"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "10", "value": {"ID": "10"}}
]
},
{
"name": "rowkey should be set when stream has double key and only id specified in insert",
"statements": [
"CREATE STREAM TEST (ID DOUBLE) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ID) VALUES (1.23);"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "1.23", "value": {"ID": 1.23}}
]
},
{
"name": "rowkey should be set when stream has bigint key and only id specified in insert",
"statements": [
"CREATE STREAM TEST (ID BIGINT) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ID) VALUES (10);"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "10", "value": {"ID": 10}}
]
},
{
"name": "rowkey should be set when stream has boolean key and only id specified in insert",
"statements": [
"CREATE STREAM TEST (ID BOOLEAN) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ID) VALUES (TRUE);"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "true", "value": {"ID": true}}
]
},
{
"name": "id should be set when stream has string key and only rowkey specicified in insert",
purplefox marked this conversation as resolved.
Show resolved Hide resolved
"statements": [
"CREATE STREAM TEST (ID STRING) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ROWKEY) VALUES ('10');"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "10", "value": {"ID": "10"}}
]
},
{
"name": "rowkey and id should match when stream has int key",
purplefox marked this conversation as resolved.
Show resolved Hide resolved
"statements": [
"CREATE STREAM TEST (ID INT) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ROWKEY, ID) VALUES ('10', 10);"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "10", "value": {"ID": 10}}
]
},
{
"name": "rowkey and id should match when stream has String key",
"statements": [
"CREATE STREAM TEST (ID STRING) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ROWKEY, ID) VALUES ('10', '10');"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "10", "value": {"ID": "10"}}
]
},
{
"name": "rowkey and id should match when stream has double key",
"statements": [
"CREATE STREAM TEST (ID DOUBLE) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ROWKEY, ID) VALUES ('1.23', 1.23);"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "1.23", "value": {"ID": 1.23}}
]
},
{
"name": "rowkey and id should match when stream has bigint key",
"statements": [
"CREATE STREAM TEST (ID BIGINT) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ROWKEY, ID) VALUES ('10', 10);"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "10", "value": {"ID": 10}}
]
},
{
"name": "rowkey and id should match when stream has boolean key",
"statements": [
"CREATE STREAM TEST (ID BOOLEAN) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ROWKEY, ID) VALUES ('true', true);"
],
"inputs": [
],
"outputs": [
{"topic": "test_topic", "key": "true", "value": {"ID": true}}
]
},
{
"name": "should fail on mismatch between explicit columns and value counts",
"statements": [
Expand All @@ -77,6 +209,18 @@
"message": "Failed to prepare statement: Expected number columns and values to match: [ROWKEY, ID], ['10']",
"status": 400
}
},
{
"name": "should fail on mismatch between rowkey and id values when stream has key",
purplefox marked this conversation as resolved.
Show resolved Hide resolved
"statements": [
"CREATE STREAM TEST (ID INT) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
"INSERT INTO TEST (ROWKEY, ID) VALUES ('10', 5);"
],
"expectedError": {
"type": "io.confluent.ksql.rest.entity.KsqlStatementErrorMessage",
"message": "Failed to insert values into 'TEST'. Expected ROWKEY and ID to match but got 10 and 5 respectively.",
"status": 400
}
}
]
}