randomdev2027 commented on issue #70429:
URL: https://github.com/apache/doris/issues/70429#issuecomment-6761927857
Full Iceberg-Definition:
```
{
"tables": {
"Ticket": {
"schema": {
"type": "struct",
"schema-id": 0,
"fields": [
{
"id": 1,
"name": "id",
"required": true,
"type": "string",
"doc": "id"
},
{
"id": 2,
"name": "issuelinks",
"required": false,
"type": {
"type": "list",
"element-id": 3,
"element-required": false,
"element": {
"type": "struct",
"fields": [
{"id": 4, "name": "relationshipType", "required": false,
"type": "string"},
{
"id": 5,
"name": "ticketRefs",
"required": false,
"type": {
"type": "list",
"element-id": 6,
"element-required": false,
"element": {
"type": "struct",
"fields": [
{"id": 7, "name": "id", "required": true, "type":
"string"}
]
}
}
}
]
}
},
"doc": "some description"
}
]
},
"partition-spec": {
"spec-id": 0,
"fields": []
},
"sort-order": {
"order-id": 0,
"fields": []
},
"stage-create": false,
"properties": {
"write.format.default": "PARQUET",
"write.metadata.delete-after-commit.enabled": "true",
"write.metadata.previous-versions-max": "100"
}
}
}
}
```
New id in this minimal example is id=7.
Full example payload:
```
{
"metadaten" : "EPMFJRAAG",
"daten" : {
"id" : "AWGU",
"ticket" : {
"typ" : "ErstelleTicket",
"id" : "PFRO",
"issuelinks" : [ {
"relationshipType" : "clones",
"ticketRefs" : [ {
"id" : "COJPDMCEC"
} ]
} ]
}
}
}
```
Full SQL:
```
SET enable_strict_cast = true;
-- MERGE: ErstelleTicketDaten und UpdateTicketDaten (Insert oder Update)
MERGE INTO
${DORIS_CATALOG}.${ICEBERG_OUTPUT_NAMESPACE}.${ICEBERG_OUTPUT_TABLE} target
USING (
SELECT
coalesce(
json_extract_string(s.payload_raw, '$.daten.ticket.id'),
json_extract_string(s.payload_raw, '$.daten.id')
) AS id,
CASE
WHEN json_extract(s.payload_raw, '$.daten.ticket.issuelinks') IS NULL
THEN NULL
ELSE CAST(
array_map(
x0 -> STRUCT(
json_extract_string(x0, '$.relationshipType'),
CAST(
array_map(
x1 -> STRUCT(
json_extract_string(x1, '$.id')
),
CAST(json_extract(x0, '$.ticketRefs') AS ARRAY<STRING>)
) AS ARRAY<STRUCT<id:STRING>>
)
),
CAST(json_extract(s.payload_raw, '$.daten.ticket.issuelinks') AS
ARRAY<STRING>)
)
AS ARRAY<STRUCT<
relationshipType:STRING,
ticketRefs:ARRAY<STRUCT<id:STRING>>
>>
)
END AS issuelinks
FROM (
SELECT s.*,
ROW_NUMBER() OVER (
PARTITION BY s.event_id
ORDER BY s.source_ts DESC, s.source_fingerprint DESC
) AS event_rank
FROM
${DORIS_STAGING_CATALOG}.${DORIS_STAGING_DATABASE}.${DORIS_STAGING_TABLE} s
WHERE s.event_type IN ('ErstelleTicketDaten', 'UpdateTicketDaten')
AND s.event_id IS NOT NULL
) s
WHERE s.event_rank = 1
) source
ON target.id = source.id
WHEN MATCHED THEN
UPDATE SET
issuelinks = source.issuelinks
WHEN NOT MATCHED THEN
INSERT (
id,
issuelinks
)
VALUES (
source.id,
source.issuelinks
);
```
We use a Staging-Table and orchestrate in this setup with Apache Airflow.
Our Payload exists in S3 and will be inserted in this staging table. Afterwards
a Airflow-Job executes this SQL. It works fine with required=false for id=7,
but fails with required=true.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
