cannot INSERT INTO Stream.

[sangli00]

detail error log:

ksql> insert into stream_merge_data_pipelinedb select * from stream_tran_to_pipeline;

io.confluent.ksql.util.KsqlException: Incompatible schema between results and sink. Result schema is [IPROBE_ID : BIGINT, ALPROTO_ID : BIGINT, USER_PORT : BIGINT, METHOD : BIGINT, USER_AGENT : VARCHAR, REQ_CONTENT_TYPE : VARCHAR, HOST : VARCHAR, REFERER : VARCHAR, VER : BIGINT, USER_IP : BIGINT, PATH : VARCHAR, URL_QUERY : VARCHAR, STATUS : BIGINT, RESP_CONTENT_LENGTH : BIGINT, REQ_CONTENT_LENGTH : BIGINT, COOKIE : VARCHAR, SERVER : VARCHAR, RESP_CTYPE : VARCHAR, REQ_TIME : BIGINT, RESP_TIME : BIGINT, DATA_STATUS : BIGINT, POLICYID : VARCHAR, REQ_HOUR : BIGINT, REGION : VARCHAR, APP_ID : BIGINT, APP_IP : BIGINT, DATA_RESOURCE_URL : VARCHAR, DATA_RESOURCE_ID : BIGINT, COUNTRY : VARCHAR, LEVEL : INT, MSH : VARCHAR], but the sink schema is [IPROBE_ID : BIGINT, ALPROTO_ID : BIGINT, USER_PORT : BIGINT, METHOD : BIGINT, USER_AGENT : VARCHAR, REQ_CONTENT_TYPE : VARCHAR, HOST : VARCHAR, REFERER : VARCHAR, VER : BIGINT, USER_IP : BIGINT, PATH : VARCHAR, URL_QUERY : VARCHAR, STATUS : BIGINT, RESP_CONTENT_LENGTH : BIGINT, REQ_CONTENT_LENGTH : BIGINT, COOKIE : VARCHAR, SERVER : VARCHAR, RESP_CTYPE : VARCHAR, REQ_TIME : BIGINT, RESP_TIME : BIGINT, DATA_STATUS : BIGINT, POLICYID : VARCHAR, REQ_HOUR : BIGINT, REGION : VARCHAR, APP_ID : BIGINT, APP_IP : BIGINT, DATA_RESOURCE_URL : VARCHAR, DATA_RESOURCE_ID : BIGINT, COUNTRY : VARCHAR, LEVEL : INT, MSH : VARCHAR].
ksql>

stream_merge_data_pipelinedb：

ksql> describe stream_merge_data_pipelinedb;

Name                 : STREAM_MERGE_DATA_PIPELINEDB
 Field               | Type
-------------------------------------------------
 ROWTIME             | BIGINT           (system)
 ROWKEY              | VARCHAR(STRING)  (system)
 IPROBE_ID           | BIGINT
 ALPROTO_ID          | BIGINT
 USER_PORT           | BIGINT
 METHOD              | BIGINT
 USER_AGENT          | VARCHAR(STRING)
 REQ_CONTENT_TYPE    | VARCHAR(STRING)
 HOST                | VARCHAR(STRING)
 REFERER             | VARCHAR(STRING)
 VER                 | BIGINT
 USER_IP             | BIGINT
 PATH                | VARCHAR(STRING)
 URL_QUERY           | VARCHAR(STRING)
 STATUS              | BIGINT
 RESP_CONTENT_LENGTH | BIGINT
 REQ_CONTENT_LENGTH  | BIGINT
 COOKIE              | VARCHAR(STRING)
 SERVER              | VARCHAR(STRING)
 RESP_CTYPE          | VARCHAR(STRING)
 REQ_TIME            | BIGINT
 RESP_TIME           | BIGINT
 DATA_STATUS         | BIGINT
 POLICYID            | VARCHAR(STRING)
 REQ_HOUR            | BIGINT
 REGION              | VARCHAR(STRING)
 APP_ID              | BIGINT
 APP_IP              | BIGINT
 DATA_RESOURCE_URL   | VARCHAR(STRING)
 DATA_RESOURCE_ID    | BIGINT
 COUNTRY             | VARCHAR(STRING)
 LEVEL               | INTEGER
 MSH                 | VARCHAR(STRING)
-------------------------------------------------
For runtime statistics and query details run: DESCRIBE EXTENDED <Stream,Table>;

stream_tran_to_pipeline:

ksql> describe stream_tran_to_pipeline;

Name                 : STREAM_TRAN_TO_PIPELINE
 Field               | Type
-------------------------------------------------
 ROWTIME             | BIGINT           (system)
 ROWKEY              | VARCHAR(STRING)  (system)
 IPROBE_ID           | BIGINT
 ALPROTO_ID          | BIGINT
 USER_PORT           | BIGINT
 METHOD              | BIGINT
 USER_AGENT          | VARCHAR(STRING)
 REQ_CONTENT_TYPE    | VARCHAR(STRING)
 HOST                | VARCHAR(STRING)
 REFERER             | VARCHAR(STRING)
 VER                 | BIGINT
 USER_IP             | BIGINT
 PATH                | VARCHAR(STRING)
 URL_QUERY           | VARCHAR(STRING)
 STATUS              | BIGINT
 RESP_CONTENT_LENGTH | BIGINT
 REQ_CONTENT_LENGTH  | BIGINT
 COOKIE              | VARCHAR(STRING)
 SERVER              | VARCHAR(STRING)
 RESP_CTYPE          | VARCHAR(STRING)
 REQ_TIME            | BIGINT
 RESP_TIME           | BIGINT
 DATA_STATUS         | BIGINT
 POLICYID            | VARCHAR(STRING)
 REQ_HOUR            | BIGINT
 REGION              | VARCHAR(STRING)
 APP_ID              | BIGINT
 APP_IP              | BIGINT
 DATA_RESOURCE_URL   | VARCHAR(STRING)
 DATA_RESOURCE_ID    | BIGINT
 COUNTRY             | VARCHAR(STRING)
 LEVEL               | INTEGER
 MSH                 | VARCHAR(STRING)
-------------------------------------------------
For runtime statistics and query details run: DESCRIBE EXTENDED <Stream,Table>;

stream stream_merge_data_pipelinedb and stream_tran_to_pipeline column&type is equal.

Why cannot insert?

[singyiu]

I run into the exact same problem, Where the schema is exactly the same but fail to insert.

Incompatible schema between results and sink. Result schema is [CREDENTIAL_ID : VARCHAR, first_name : VARCHAR, last_name : VARCHAR, access_card_id_binary : VARCHAR, is_active : BOOLEAN, effective_date : VARCHAR, expiry_date : VARCHAR, access_levels : ARRAY], but the sink schema is [CREDENTIAL_ID : VARCHAR, first_name : VARCHAR, last_name : VARCHAR, access_card_id_binary : VARCHAR, is_active : BOOLEAN, effective_date : VARCHAR, expiry_date : VARCHAR, access_levels : ARRAY].

[big-andy-coates]

Humm... this is worrying. Are you able to provide steps to recreate please, i.e. the create statements and the INSERT statement in question? Then we can look into it.

[big-andy-coates]

I tried adding a test case to insert-into.json:

    {
      "name": "complex schema",
      "statements": [
        "CREATE STREAM SOURCE (IPROBE_ID BIGINT, ALPROTO_ID BIGINT, USER_PORT BIGINT, METHOD BIGINT, USER_AGENT STRING, REQ_CONTENT_TYPE STRING, HOST STRING, REFERER STRING, VER BIGINT, USER_IP BIGINT, PATH STRING, URL_QUERY STRING, STATUS BIGINT, RESP_CONTENT_LENGTH BIGINT, REQ_CONTENT_LENGTH BIGINT, COOKIE STRING, SERVER STRING, RESP_CTYPE STRING, REQ_TIME BIGINT, RESP_TIME BIGINT, DATA_STATUS BIGINT, POLICYID STRING, REQ_HOUR BIGINT, REGION STRING, APP_ID BIGINT, APP_IP BIGINT, DATA_RESOURCE_URL STRING, DATA_RESOURCE_ID BIGINT, COUNTRY STRING, LEVEL INTEGER, MSH STRING) WITH (kafka_topic='source', value_format='JSON');",
        "CREATE STREAM SINK   (IPROBE_ID BIGINT, ALPROTO_ID BIGINT, USER_PORT BIGINT, METHOD BIGINT, USER_AGENT STRING, REQ_CONTENT_TYPE STRING, HOST STRING, REFERER STRING, VER BIGINT, USER_IP BIGINT, PATH STRING, URL_QUERY STRING, STATUS BIGINT, RESP_CONTENT_LENGTH BIGINT, REQ_CONTENT_LENGTH BIGINT, COOKIE STRING, SERVER STRING, RESP_CTYPE STRING, REQ_TIME BIGINT, RESP_TIME BIGINT, DATA_STATUS BIGINT, POLICYID STRING, REQ_HOUR BIGINT, REGION STRING, APP_ID BIGINT, APP_IP BIGINT, DATA_RESOURCE_URL STRING, DATA_RESOURCE_ID BIGINT, COUNTRY STRING, LEVEL INTEGER, MSH STRING) WITH (kafka_topic='sink', value_format='JSON');",
        "INSERT INTO SINK SELECT * FROM SOURCE;"
      ],
      "inputs": [
        {"topic": "source", "key": 0, "value": null, "timestamp": 0}
      ],
      "outputs": [
        {"topic": "sink", "key": 0, "value": null, "timestamp": 0}
      ]
    }

This test did not fail with the error you encountered, so I'm unable to recreate. This may be some quirk of how your source and sink streams are being created. Hence requiring steps to recreate to look into this.

Thanks,

Andy

[aurabhi]

Got into a similar problem. selecting a subset of values and insert into another stream which has the same schema.

insert into sub_stream select NAME, LASTNAME, CREATED, LOCATION from main_stream emit changes;

Caused by: io.confluent.ksql.api.client.exception.KsqlClientException: Received 400 response from server: Incompatible schema between results and sink.
Result schema is NAME STRING, LASTNAME STRING, CREATED BIGINT, LOCATION STRING
Sink schema is NAME STRING, LASTNAME STRING, CREATED BIGINT, LOCATION STRING, EMAIL STRING, ADDRESS MAP<STRING, STRING>, PHONE STRING.

Error code: 40001

[vcrfxia]

Hi @aurabhi , the behavior you reported is expected: ksqlDB requires schemas for INSERT INTO statements to match exactly. Because the schema you are trying to insert is a subset of the required schema, you can modify your statement to meet the requirements by adding in null-valued columns for the missing columns:

insert into sub_stream select NAME, LASTNAME, CREATED, LOCATION, cast(null as varchar) as email, cast(null as map<string, string>) as address, cast(null as varchar) as phone from main_stream emit changes;

I'm going to close this ticket for now as there is a workaround in your case and the original question has not seen activity in more than a year. Feel free to reopen if this has not addressed your question. Thanks.