Skip to content

Fix kafka_fdw unexpected errors #3

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

Merged
my-ship-it merged 1 commit into from
Nov 6, 2025
Merged

Fix kafka_fdw unexpected errors #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions src/kafka_fdw.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1271,6 +1271,7 @@ kafkaBeginForeignModify(ModifyTableState *mtstate,
Relation rel = rinfo->ri_RelationDesc;
char errstr[512]; /* librdkafka API error reporting buffer */
List *attnumlist;
KafkaPartitionList *partition_list;

DEBUGLOG("%s", __func__);

Expand Down Expand Up @@ -1376,6 +1377,10 @@ kafkaBeginForeignModify(ModifyTableState *mtstate,
festate->kafka_topic_handle = rkt;
festate->kafka_handle = rk;

partition_list = getPartitionList(festate->kafka_handle,
festate->kafka_topic_handle);
festate->partition_cnt = partition_list->partition_cnt;

rinfo->ri_FdwState = festate;
}

Expand Down Expand Up @@ -1427,6 +1432,15 @@ kafkaExecForeignInsert(EState *estate, ResultRelInfo *rinfo, TupleTableSlot *slo
if (!isnull)
partition = DatumGetInt32(value);

if (!isnull && partition >= festate->partition_cnt)
{
ereport(ERROR, (errcode(ERRCODE_FDW_ERROR),
errmsg("invalid partition number %d", partition),
errdetail("Topic \"%s\" has only %d partitions (0-%d).",
rd_kafka_topic_name(festate->kafka_topic_handle),
festate->partition_cnt, festate->partition_cnt - 1)));
}

DEBUGLOG("Message: %s", festate->attribute_buf.data);

retry:
Expand Down
2 changes: 1 addition & 1 deletion src/kafka_fdw.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,7 +246,7 @@ typedef struct KafkaFdwModifyState
List * partition_list; /* integer list of partitions */
StringInfoData attname_buf; /* buffer holding attribute names for json format */
char ** attnames; /* pointer into attname_buf */

int32 partition_cnt;
} KafkaFdwModifyState;
/* connection.c */
void KafkaFdwGetConnection(KafkaOptions *k_options,
Expand Down
18 changes: 18 additions & 0 deletions src/option.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -487,6 +487,24 @@ get_kafka_fdw_attribute_options(Oid relid, KafkaOptions *kafka_options)
}
}

if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
{
if (kafka_options->partition_attnum == -1)
{
ereport(ERROR,
(errcode(ERRCODE_FDW_ERROR),
errmsg("missing column definition with option (partition 'true')"),
errhint("Specify a column and set its option to (partition 'true')")));
}
if (kafka_options->offset_attnum == -1)
{
ereport(ERROR,
(errcode(ERRCODE_FDW_ERROR),
errmsg("missing column definition with option (offset 'true')"),
errhint("Specify a column and set its option to (offset 'true')")));
}
}

/* calculate number of parsable columns */

if (kafka_options->partition_attnum != -1)
Expand Down
54 changes: 54 additions & 0 deletions test/expected/cbdb_test.out
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
\i test/sql/setup.inc
\set ECHO none
-- This file contains extra tests for Apache Cloudberry
-- 1.column options 'partition' and 'offset ' are essential
CREATE FOREIGN TABLE invalid_kafka_table (
some_int int OPTIONS (json 'int_val'),
some_text text OPTIONS (json 'text_val'),
some_date date OPTIONS (json 'date_val'),
some_time timestamp OPTIONS (json 'time_val')
)
SERVER kafka_server OPTIONS
(format 'json', topic 'contrib_regress_cloudberry', batch_size '30', buffer_delay '500');
select * from invalid_kafka_table;
ERROR: missing column definition with option (partition 'true')
HINT: Specify a column and set its option to (partition 'true')
drop foreign table invalid_kafka_table;
CREATE FOREIGN TABLE invalid_kafka_table (
part int OPTIONS (partition 'true'),
offs bigint OPTIONS (offset 'false'),
some_int int OPTIONS (json 'int_val'),
some_text text OPTIONS (json 'text_val'),
some_date date OPTIONS (json 'date_val'),
some_time timestamp OPTIONS (json 'time_val')
)
SERVER kafka_server OPTIONS
(format 'json', topic 'contrib_regress_cloudberry', batch_size '30', buffer_delay '500');
select * from invalid_kafka_table;
ERROR: missing column definition with option (offset 'true')
HINT: Specify a column and set its option to (offset 'true')
drop foreign table invalid_kafka_table;
-- 2. insert into a non-exist partition
CREATE FOREIGN TABLE normal_kafka_table (
part int OPTIONS (partition 'true'),
offs bigint OPTIONS (offset 'true'),
some_int int OPTIONS (json 'int_val'),
some_text text OPTIONS (json 'text_val'),
some_date date OPTIONS (json 'date_val'),
some_time timestamp OPTIONS (json 'time_val')
)
SERVER kafka_server OPTIONS
(format 'json', topic 'contrib_regress_cloudberry', batch_size '30', buffer_delay '500');
-- invalid partition
insert into normal_kafka_table(part, some_int, some_text, some_date, some_time) values (10, 1, 'jack', now(), now());
ERROR: invalid partition number 10 (seg2 172.17.0.3:7004 pid=24750)
DETAIL: Topic "contrib_regress_cloudberry" has only 3 partitions (0-2).
-- valid partition
insert into normal_kafka_table(part, some_int, some_text, some_date, some_time) values (0, 1, 'jack', now(), now());
select count(*) from normal_kafka_table;
count
-------
2
(1 row)

drop foreign table normal_kafka_table;
20 changes: 10 additions & 10 deletions test/expected/kafka_test.out
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,8 +243,8 @@ SELECT explain_invariant('SELECT * FROM kafka_test_part WHERE offs = (SELECT 1)
(11 rows)

SELECT explain_invariant('SELECT * FROM kafka_test_part WHERE offs = 0 AND part = (SELECT 1)');
explain_invariant
-------------------------------------------------------------------------------------
explain_invariant
---------------------------------------------------------------------------------------
Gather Motion 3:1 (slice1; segments: 3) (actual rows=1 loops=1)
InitPlan 1 (returns $0) (slice2)
-> Result (actual rows=1 loops=1)
Expand All @@ -259,8 +259,8 @@ SELECT explain_invariant('SELECT * FROM kafka_test_part WHERE offs = 0 AND part
(11 rows)

SELECT explain_invariant('SELECT * FROM kafka_test_part WHERE offs < (SELECT 1) AND part = 0');
explain_invariant
-------------------------------------------------------------------------------------
explain_invariant
---------------------------------------------------------------------------------------
Gather Motion 3:1 (slice1; segments: 3) (actual rows=1 loops=1)
InitPlan 1 (returns $0) (slice2)
-> Result (actual rows=1 loops=1)
Expand All @@ -275,8 +275,8 @@ SELECT explain_invariant('SELECT * FROM kafka_test_part WHERE offs < (SELECT 1)
(11 rows)

SELECT explain_invariant('SELECT * FROM kafka_test_part WHERE offs <= (SELECT 1) AND part = 0');
explain_invariant
-------------------------------------------------------------------------------------
explain_invariant
---------------------------------------------------------------------------------------
Gather Motion 3:1 (slice1; segments: 3) (actual rows=2 loops=1)
InitPlan 1 (returns $0) (slice2)
-> Result (actual rows=1 loops=1)
Expand All @@ -291,8 +291,8 @@ SELECT explain_invariant('SELECT * FROM kafka_test_part WHERE offs <= (SELECT 1)
(11 rows)

SELECT explain_invariant('SELECT * FROM kafka_test_part WHERE offs = 0 AND part < (SELECT 1)');
explain_invariant
-------------------------------------------------------------------------------------
explain_invariant
---------------------------------------------------------------------------------------
Gather Motion 3:1 (slice1; segments: 3) (actual rows=1 loops=1)
InitPlan 1 (returns $0) (slice2)
-> Result (actual rows=1 loops=1)
Expand All @@ -307,8 +307,8 @@ SELECT explain_invariant('SELECT * FROM kafka_test_part WHERE offs = 0 AND part
(11 rows)

SELECT explain_invariant('SELECT * FROM kafka_test_part WHERE offs = 0 AND part <= (SELECT 1)');
explain_invariant
-------------------------------------------------------------------------------------
explain_invariant
---------------------------------------------------------------------------------------
Gather Motion 3:1 (slice1; segments: 3) (actual rows=2 loops=1)
InitPlan 1 (returns $0) (slice2)
-> Result (actual rows=1 loops=1)
Expand Down
8 changes: 6 additions & 2 deletions test/init_kafka.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@ KAFKA_BIN_DIR=~/kafka
: ${KAFKA_PRODUCER:="${KAFKA_BIN_DIR}/bin/kafka-console-producer.sh"}
: ${KAFKA_TOPICS:="${KAFKA_BIN_DIR}/bin/kafka-topics.sh"}

topics=( contrib_regress4 contrib_regress contrib_regress_prod contrib_regress_prod_json contrib_regress_junk contrib_regress_json contrib_regress_json_junk )
partitions=( 4 1 4 4 1 1 1 )
topics=( contrib_regress4 contrib_regress contrib_regress_prod contrib_regress_prod_json contrib_regress_junk contrib_regress_json contrib_regress_json_junk contrib_regress_cloudberry)
partitions=( 4 1 4 4 1 1 1 3 )

if [[ -n "${PG_CONFIG}" ]]; then
BIN=$(${PG_CONFIG} --bindir)
Expand Down Expand Up @@ -75,3 +75,7 @@ $kafka_cmd contrib_regress_json_junk <<-EOF
{"int_val" : 999851, "text_val" : "correct line null time", "date_val" : "2015-01-12", "time_val" : null}
{"int_val" : 999871, "invalid json no time" : "invalid json", "date_val" : "2015-01-12", "time_val" : }
EOF

$kafka_cmd contrib_regress_cloudberry <<-EOF
{"int_val" : 8893920, "text_val" : "correct line", "date_val" : "2015-01-12", "time_val" : "2015-01-12T13:42:21"}
EOF
4 changes: 3 additions & 1 deletion test/run_kafka.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,9 @@ DIST=$(cat /etc/os-release | grep ^ID= | sed s/ID=//)
echo

# # Download Apache Kafka
wget https://downloads.apache.org/kafka/${KAFKA_VERSION}/${KAFKA_ARCHIVE}
if [ ! -f "${KAFKA_ARCHIVE}" ]; then
wget https://downloads.apache.org/kafka/${KAFKA_VERSION}/${KAFKA_ARCHIVE}
fi
tar -xzf ${KAFKA_ARCHIVE} -C ${KAFKA_BIN_DIR} --strip-components=1
export PATH="${KAFKA_BIN_DIR}/bin/:$PATH"

Expand Down
53 changes: 53 additions & 0 deletions test/sql/cbdb_test.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
\i test/sql/setup.inc

-- This file contains extra tests for Apache Cloudberry

-- 1.column options 'partition' and 'offset ' are essential
CREATE FOREIGN TABLE invalid_kafka_table (
some_int int OPTIONS (json 'int_val'),
some_text text OPTIONS (json 'text_val'),
some_date date OPTIONS (json 'date_val'),
some_time timestamp OPTIONS (json 'time_val')
)
SERVER kafka_server OPTIONS
(format 'json', topic 'contrib_regress_cloudberry', batch_size '30', buffer_delay '500');

select * from invalid_kafka_table;

drop foreign table invalid_kafka_table;

CREATE FOREIGN TABLE invalid_kafka_table (
part int OPTIONS (partition 'true'),
offs bigint OPTIONS (offset 'false'),
some_int int OPTIONS (json 'int_val'),
some_text text OPTIONS (json 'text_val'),
some_date date OPTIONS (json 'date_val'),
some_time timestamp OPTIONS (json 'time_val')
)
SERVER kafka_server OPTIONS
(format 'json', topic 'contrib_regress_cloudberry', batch_size '30', buffer_delay '500');

select * from invalid_kafka_table;
drop foreign table invalid_kafka_table;

-- 2. insert into a non-exist partition
CREATE FOREIGN TABLE normal_kafka_table (
part int OPTIONS (partition 'true'),
offs bigint OPTIONS (offset 'true'),
some_int int OPTIONS (json 'int_val'),
some_text text OPTIONS (json 'text_val'),
some_date date OPTIONS (json 'date_val'),
some_time timestamp OPTIONS (json 'time_val')
)
SERVER kafka_server OPTIONS
(format 'json', topic 'contrib_regress_cloudberry', batch_size '30', buffer_delay '500');

-- invalid partition
insert into normal_kafka_table(part, some_int, some_text, some_date, some_time) values (10, 1, 'jack', now(), now());

-- valid partition
insert into normal_kafka_table(part, some_int, some_text, some_date, some_time) values (0, 1, 'jack', now(), now());

select count(*) from normal_kafka_table;

drop foreign table normal_kafka_table;
Loading