This is an automated email from the ASF dual-hosted git repository.
ostinru pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cloudberry-pxf.git
The following commit(s) were added to refs/heads/main by this push:
new 3b5dead7 Observability: pxf_stat_activity and pxf_cancel_backend (#124)
3b5dead7 is described below
commit 3b5dead7146b8fea280c5e596c8bf52e9efeaaa9
Author: Nikolay Antonov <[email protected]>
AuthorDate: Sat Aug 29 16:17:58 2026 +0500
Observability: pxf_stat_activity and pxf_cancel_backend (#124)
### Add `pxf_stat_activity`
Add `pxf_stat_activity` to help DBAs find out what happening inside PXF
server.
```
postgres=# select * from pxf_stat_activity ;
-[ RECORD 1 ]-+---------------------------------------------------------
segment_id | 1
session_id | 57997
command_count | 8
xid | 57997-0000000008
operation | READ_BRIDGE
usename | gpadmin
server | ostinru-test
profile | s3:parquet
schema_name | public
table_name | test_blob_read
data_source | ostinru-test/topics/cuscom.email.created.v1/2026-05-29/*
query_start | 2026-06-23 16:16:48.318+03
pxf_host | example.com
```
### Add `pxf_cancel_backend` and `pxf_interrupt_backend`
Add `pxf_cancel_backend` and `pxf_interrupt_backend` to allow killing
running queries.
### Update tests
1. Update pxfExtension tests - use testconatiners.
2. Explicitly define extension version. So, there is no need in sql file
manipulations.
---
.../create_extension/expected/query01.ans | 91 ++++++++
.../create_extension/sql/query01.sql | 55 +++++
.../step_1_create_2_1/expected/query01.ans | 64 ++++++
.../step_1_create_2_1/sql/query01.sql | 46 ++++
.../step_2_after_upgrade_2_2/expected/query01.ans | 91 ++++++++
.../step_2_after_upgrade_2_2/sql/query01.sql | 55 +++++
.../expected/query01.ans | 64 ++++++
.../step_3_after_downgrade_2_1/sql/query01.sql | 46 ++++
.../features/extension/PxfExtensionTest.java | 18 ++
external-table/Makefile | 4 +-
external-table/pxf--2.1--2.2.sql | 99 ++++++++
external-table/pxf--2.2--2.1.sql | 39 ++++
external-table/pxf--2.2.sql | 130 +++++++++++
external-table/pxf.control | 2 +-
external-table/src/pxf_cancel_activity.c | 173 ++++++++++++++
external-table/src/pxf_stat_activity.c | 142 ++++++++++++
.../pxf/service/activity/ActiveRequest.java | 167 ++++++++++++++
.../pxf/service/activity/ActiveRequestInfo.java | 96 ++++++++
.../service/activity/ActiveRequestRegistry.java | 191 ++++++++++++++++
.../pxf/service/controller/BaseServiceImpl.java | 52 ++++-
.../pxf/service/controller/ReadServiceImpl.java | 30 ++-
.../pxf/service/controller/WriteServiceImpl.java | 27 ++-
.../service/rest/PxfBackendControlResource.java | 83 +++++++
.../pxf/service/rest/PxfStatActivityResource.java | 66 ++++++
.../activity/ActiveRequestRegistryTest.java | 250 +++++++++++++++++++++
.../service/controller/ReadServiceImplTest.java | 3 +-
.../service/controller/WriteServiceImplTest.java | 3 +-
.../service/rest/PxfBackendControlResourceIT.java | 92 ++++++++
.../service/rest/PxfStatActivityResourceIT.java | 89 ++++++++
29 files changed, 2242 insertions(+), 26 deletions(-)
diff --git
a/automation/sqlrepo/features/extension_tests_2_2/create_extension/expected/query01.ans
b/automation/sqlrepo/features/extension_tests_2_2/create_extension/expected/query01.ans
new file mode 100644
index 00000000..72792778
--- /dev/null
+++
b/automation/sqlrepo/features/extension_tests_2_2/create_extension/expected/query01.ans
@@ -0,0 +1,91 @@
+-- start_ignore
+-- end_ignore
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- @description query01 for PXF 2.2 install test (pxf_stat_activity + backend
control)
+-- start_matchsubs
+--
+-- m{.*/usr/local/pxf-(dev|gp\d).*}
+-- s{/usr/local/pxf-(dev|gp\d)}{\$PXF_HOME}
+--
+-- m{.*\$libdir/pxf.*}
+-- s{\$libdir}{\$PXF_HOME/gpextable}
+--
+-- end_matchsubs
+-- start_ignore
+\c pxfautomation_extension
+-- end_ignore
+
+SELECT extversion FROM pg_extension WHERE extname = 'pxf';
+ extversion
+------------
+ 2.2
+(1 row)
+
+SELECT p.proname
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_proc AS p ON (p.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+ proname
+---------------------------
+ pxf_cancel_backend
+ pxf_cancel_backend_raw
+ pxf_interrupt_backend
+ pxf_interrupt_backend_raw
+ pxf_read
+ pxf_stat_activity_raw
+ pxf_validate
+ pxf_write
+ pxfdelimited_import
+ pxfwritable_export
+ pxfwritable_import
+(11 rows)
+
+SELECT c.relname, c.relkind
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_class AS c ON (c.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+ relname | relkind
+-------------------+---------
+ pxf_stat_activity | v
+(1 row)
+
+-- the backend-control functions are reachable end-to-end (C -> local PXF ->
registry);
+-- with no in-flight requests they are no-ops returning 0
+SELECT pxf_cancel_backend(-1);
+ pxf_cancel_backend
+--------------------
+ 0
+(1 row)
+
+SELECT pxf_interrupt_backend(-1);
+ pxf_interrupt_backend
+-----------------------
+ 0
+(1 row)
+
+-- the activity view is empty while PXF is idle
+SELECT count(*) FROM pxf_stat_activity;
+ count
+-------
+ 0
+(1 row)
diff --git
a/automation/sqlrepo/features/extension_tests_2_2/create_extension/sql/query01.sql
b/automation/sqlrepo/features/extension_tests_2_2/create_extension/sql/query01.sql
new file mode 100644
index 00000000..dc067126
--- /dev/null
+++
b/automation/sqlrepo/features/extension_tests_2_2/create_extension/sql/query01.sql
@@ -0,0 +1,55 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- @description query01 for PXF 2.2 install test (pxf_stat_activity + backend
control)
+-- start_matchsubs
+--
+-- m{.*/usr/local/pxf-(dev|gp\d).*}
+-- s{/usr/local/pxf-(dev|gp\d)}{\$PXF_HOME}
+--
+-- m{.*\$libdir/pxf.*}
+-- s{\$libdir}{\$PXF_HOME/gpextable}
+--
+-- end_matchsubs
+-- start_ignore
+\c pxfautomation_extension
+-- end_ignore
+
+SELECT extversion FROM pg_extension WHERE extname = 'pxf';
+
+SELECT p.proname
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_proc AS p ON (p.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+
+SELECT c.relname, c.relkind
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_class AS c ON (c.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+
+-- the backend-control functions are reachable end-to-end (C -> local PXF ->
registry);
+-- with no in-flight requests they are no-ops returning 0
+SELECT pxf_cancel_backend(-1);
+
+SELECT pxf_interrupt_backend(-1);
+
+-- the activity view is empty while PXF is idle
+SELECT count(*) FROM pxf_stat_activity;
diff --git
a/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_1_create_2_1/expected/query01.ans
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_1_create_2_1/expected/query01.ans
new file mode 100644
index 00000000..b1f54f67
--- /dev/null
+++
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_1_create_2_1/expected/query01.ans
@@ -0,0 +1,64 @@
+-- start_ignore
+-- end_ignore
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- @description query01 for PXF 2.2 upgrade/downgrade test - step 1: create at
2.1
+-- start_matchsubs
+--
+-- m{.*/usr/local/pxf-(dev|gp\d).*}
+-- s{/usr/local/pxf-(dev|gp\d)}{\$PXF_HOME}
+--
+-- m{.*\$libdir/pxf.*}
+-- s{\$libdir}{\$PXF_HOME/gpextable}
+--
+-- end_matchsubs
+-- start_ignore
+\c pxfautomation_extension
+-- end_ignore
+
+SELECT extversion FROM pg_extension WHERE extname = 'pxf';
+ extversion
+------------
+ 2.1
+(1 row)
+
+SELECT p.proname
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_proc AS p ON (p.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+ proname
+---------------------
+ pxf_read
+ pxf_validate
+ pxf_write
+ pxfdelimited_import
+ pxfwritable_export
+ pxfwritable_import
+(6 rows)
+
+SELECT c.relname, c.relkind
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_class AS c ON (c.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+ relname | relkind
+---------+---------
+(0 rows)
diff --git
a/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_1_create_2_1/sql/query01.sql
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_1_create_2_1/sql/query01.sql
new file mode 100644
index 00000000..c708069b
--- /dev/null
+++
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_1_create_2_1/sql/query01.sql
@@ -0,0 +1,46 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- @description query01 for PXF 2.2 upgrade/downgrade test - step 1: create at
2.1
+-- start_matchsubs
+--
+-- m{.*/usr/local/pxf-(dev|gp\d).*}
+-- s{/usr/local/pxf-(dev|gp\d)}{\$PXF_HOME}
+--
+-- m{.*\$libdir/pxf.*}
+-- s{\$libdir}{\$PXF_HOME/gpextable}
+--
+-- end_matchsubs
+-- start_ignore
+\c pxfautomation_extension
+-- end_ignore
+
+SELECT extversion FROM pg_extension WHERE extname = 'pxf';
+
+SELECT p.proname
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_proc AS p ON (p.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+
+SELECT c.relname, c.relkind
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_class AS c ON (c.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
diff --git
a/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_2_after_upgrade_2_2/expected/query01.ans
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_2_after_upgrade_2_2/expected/query01.ans
new file mode 100644
index 00000000..4d1c0a1c
--- /dev/null
+++
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_2_after_upgrade_2_2/expected/query01.ans
@@ -0,0 +1,91 @@
+-- start_ignore
+-- end_ignore
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- @description query01 for PXF 2.2 upgrade/downgrade test - step 2: after
upgrade to 2.2
+-- start_matchsubs
+--
+-- m{.*/usr/local/pxf-(dev|gp\d).*}
+-- s{/usr/local/pxf-(dev|gp\d)}{\$PXF_HOME}
+--
+-- m{.*\$libdir/pxf.*}
+-- s{\$libdir}{\$PXF_HOME/gpextable}
+--
+-- end_matchsubs
+-- start_ignore
+\c pxfautomation_extension
+-- end_ignore
+
+SELECT extversion FROM pg_extension WHERE extname = 'pxf';
+ extversion
+------------
+ 2.2
+(1 row)
+
+SELECT p.proname
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_proc AS p ON (p.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+ proname
+---------------------------
+ pxf_cancel_backend
+ pxf_cancel_backend_raw
+ pxf_interrupt_backend
+ pxf_interrupt_backend_raw
+ pxf_read
+ pxf_stat_activity_raw
+ pxf_validate
+ pxf_write
+ pxfdelimited_import
+ pxfwritable_export
+ pxfwritable_import
+(11 rows)
+
+SELECT c.relname, c.relkind
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_class AS c ON (c.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+ relname | relkind
+-------------------+---------
+ pxf_stat_activity | v
+(1 row)
+
+-- the backend-control functions are reachable end-to-end (C -> local PXF ->
registry);
+-- with no in-flight requests they are no-ops returning 0
+SELECT pxf_cancel_backend(-1);
+ pxf_cancel_backend
+--------------------
+ 0
+(1 row)
+
+SELECT pxf_interrupt_backend(-1);
+ pxf_interrupt_backend
+-----------------------
+ 0
+(1 row)
+
+-- the activity view is empty while PXF is idle
+SELECT count(*) FROM pxf_stat_activity;
+ count
+-------
+ 0
+(1 row)
diff --git
a/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_2_after_upgrade_2_2/sql/query01.sql
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_2_after_upgrade_2_2/sql/query01.sql
new file mode 100644
index 00000000..f83ae490
--- /dev/null
+++
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_2_after_upgrade_2_2/sql/query01.sql
@@ -0,0 +1,55 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- @description query01 for PXF 2.2 upgrade/downgrade test - step 2: after
upgrade to 2.2
+-- start_matchsubs
+--
+-- m{.*/usr/local/pxf-(dev|gp\d).*}
+-- s{/usr/local/pxf-(dev|gp\d)}{\$PXF_HOME}
+--
+-- m{.*\$libdir/pxf.*}
+-- s{\$libdir}{\$PXF_HOME/gpextable}
+--
+-- end_matchsubs
+-- start_ignore
+\c pxfautomation_extension
+-- end_ignore
+
+SELECT extversion FROM pg_extension WHERE extname = 'pxf';
+
+SELECT p.proname
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_proc AS p ON (p.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+
+SELECT c.relname, c.relkind
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_class AS c ON (c.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+
+-- the backend-control functions are reachable end-to-end (C -> local PXF ->
registry);
+-- with no in-flight requests they are no-ops returning 0
+SELECT pxf_cancel_backend(-1);
+
+SELECT pxf_interrupt_backend(-1);
+
+-- the activity view is empty while PXF is idle
+SELECT count(*) FROM pxf_stat_activity;
diff --git
a/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_3_after_downgrade_2_1/expected/query01.ans
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_3_after_downgrade_2_1/expected/query01.ans
new file mode 100644
index 00000000..f91bb17b
--- /dev/null
+++
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_3_after_downgrade_2_1/expected/query01.ans
@@ -0,0 +1,64 @@
+-- start_ignore
+-- end_ignore
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- @description query01 for PXF 2.2 upgrade/downgrade test - step 3: after
downgrade to 2.1
+-- start_matchsubs
+--
+-- m{.*/usr/local/pxf-(dev|gp\d).*}
+-- s{/usr/local/pxf-(dev|gp\d)}{\$PXF_HOME}
+--
+-- m{.*\$libdir/pxf.*}
+-- s{\$libdir}{\$PXF_HOME/gpextable}
+--
+-- end_matchsubs
+-- start_ignore
+\c pxfautomation_extension
+-- end_ignore
+
+SELECT extversion FROM pg_extension WHERE extname = 'pxf';
+ extversion
+------------
+ 2.1
+(1 row)
+
+SELECT p.proname
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_proc AS p ON (p.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+ proname
+---------------------
+ pxf_read
+ pxf_validate
+ pxf_write
+ pxfdelimited_import
+ pxfwritable_export
+ pxfwritable_import
+(6 rows)
+
+SELECT c.relname, c.relkind
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_class AS c ON (c.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+ relname | relkind
+---------+---------
+(0 rows)
diff --git
a/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_3_after_downgrade_2_1/sql/query01.sql
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_3_after_downgrade_2_1/sql/query01.sql
new file mode 100644
index 00000000..6f28f334
--- /dev/null
+++
b/automation/sqlrepo/features/extension_tests_2_2/upgrade_downgrade/step_3_after_downgrade_2_1/sql/query01.sql
@@ -0,0 +1,46 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- @description query01 for PXF 2.2 upgrade/downgrade test - step 3: after
downgrade to 2.1
+-- start_matchsubs
+--
+-- m{.*/usr/local/pxf-(dev|gp\d).*}
+-- s{/usr/local/pxf-(dev|gp\d)}{\$PXF_HOME}
+--
+-- m{.*\$libdir/pxf.*}
+-- s{\$libdir}{\$PXF_HOME/gpextable}
+--
+-- end_matchsubs
+-- start_ignore
+\c pxfautomation_extension
+-- end_ignore
+
+SELECT extversion FROM pg_extension WHERE extname = 'pxf';
+
+SELECT p.proname
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_proc AS p ON (p.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
+
+SELECT c.relname, c.relkind
+FROM pg_catalog.pg_extension AS e
+ INNER JOIN pg_catalog.pg_depend AS d ON (d.refobjid = e.oid)
+ INNER JOIN pg_catalog.pg_class AS c ON (c.oid = d.objid)
+WHERE d.deptype = 'e' AND e.extname = 'pxf'
+ORDER BY 1;
diff --git
a/automation/src/test/java/org/apache/cloudberry/pxf/automation/features/extension/PxfExtensionTest.java
b/automation/src/test/java/org/apache/cloudberry/pxf/automation/features/extension/PxfExtensionTest.java
index 77cbaaa3..4ebf6c4a 100644
---
a/automation/src/test/java/org/apache/cloudberry/pxf/automation/features/extension/PxfExtensionTest.java
+++
b/automation/src/test/java/org/apache/cloudberry/pxf/automation/features/extension/PxfExtensionTest.java
@@ -100,4 +100,22 @@ public class PxfExtensionTest extends
AbstractTestcontainersTest {
extensionDb.runQuery("ALTER EXTENSION pxf UPDATE TO '2.1'");
regress.runSqlTest("features/extension_tests/downgrade_then_upgrade/step_3_after_alter_extension_upgrade");
}
+
+ @Test(groups = {"testcontainers", "pxf-extension"})
+ public void testPxfCreateExtension22() throws Exception {
+ extensionDb.runQuery("CREATE EXTENSION pxf VERSION '2.2'");
+ regress.runSqlTest("features/extension_tests_2_2/create_extension");
+ }
+
+ @Test(groups = {"testcontainers", "pxf-extension"})
+ public void testPxfUpgradeDowngrade22() throws Exception {
+ extensionDb.runQuery("CREATE EXTENSION pxf VERSION '2.1'");
+
regress.runSqlTest("features/extension_tests_2_2/upgrade_downgrade/step_1_create_2_1");
+
+ extensionDb.runQuery("ALTER EXTENSION pxf UPDATE TO '2.2'");
+
regress.runSqlTest("features/extension_tests_2_2/upgrade_downgrade/step_2_after_upgrade_2_2");
+
+ extensionDb.runQuery("ALTER EXTENSION pxf UPDATE TO '2.1'");
+
regress.runSqlTest("features/extension_tests_2_2/upgrade_downgrade/step_3_after_downgrade_2_1");
+ }
}
diff --git a/external-table/Makefile b/external-table/Makefile
index ae5195dc..ebead67b 100644
--- a/external-table/Makefile
+++ b/external-table/Makefile
@@ -1,7 +1,7 @@
EXTENSION = pxf
-DATA = pxf--2.1.sql pxf--2.0--2.1.sql pxf--2.1--2.0.sql pxf--2.0.sql
pxf--1.0--2.0.sql pxf--1.0.sql
+DATA = pxf--2.2.sql pxf--2.1--2.2.sql pxf--2.2--2.1.sql pxf--2.1.sql
pxf--2.0--2.1.sql pxf--2.1--2.0.sql pxf--2.0.sql pxf--1.0--2.0.sql pxf--1.0.sql
MODULE_big = pxf
-OBJS = src/pxfprotocol.o src/pxfbridge.o src/pxfuriparser.o
src/libchurl.o src/pxfutils.o src/pxfheaders.o src/gpdbwritableformatter.o
src/pxfdelimited_formatter.o src/pxffilters.o
+OBJS = src/pxfprotocol.o src/pxfbridge.o src/pxfuriparser.o
src/libchurl.o src/pxfutils.o src/pxfheaders.o src/gpdbwritableformatter.o
src/pxfdelimited_formatter.o src/pxffilters.o src/pxf_stat_activity.o
src/pxf_cancel_activity.o
REGRESS = setup pxf pxfinvalid
SHLIB_LINK += -lcurl
diff --git a/external-table/pxf--2.1--2.2.sql b/external-table/pxf--2.1--2.2.sql
new file mode 100644
index 00000000..bdfd87a7
--- /dev/null
+++ b/external-table/pxf--2.1--2.2.sql
@@ -0,0 +1,99 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+/* external_table/pxf--2.1--2.2.sql */
+
+------------------------------------------------------------------
+-- PXF Activity Monitoring
+------------------------------------------------------------------
+
+-- Raw per-segment accessor: each segment asks its local PXF instance for the
+-- activity that originates from its own segment id and returns the JSON body
+-- verbatim as a single row. Dispatched to every segment; the typed columns are
+-- produced by the pxf_stat_activity view below. The function is set-returning
+-- (one row per segment) because EXECUTE ON ALL SEGMENTS is only permitted for
+-- set-returning functions.
+CREATE FUNCTION pxf_stat_activity_raw() RETURNS SETOF text
+AS 'MODULE_PATHNAME', 'pxf_stat_activity_raw'
+LANGUAGE C VOLATILE EXECUTE ON ALL SEGMENTS;
+
+-- pg_stat_activity-like view of the queries currently running inside PXF,
+-- aggregated across all segment hosts. DISTINCT is a safety net; PXF already
+-- de-duplicates by filtering each response to the requesting segment id.
+CREATE VIEW pxf_stat_activity AS
+SELECT DISTINCT
+ (a->>'segmentId')::int AS segment_id,
+ (a->>'gpSessionId')::int AS session_id,
+ (a->>'gpCommandCount')::int AS command_count,
+ a->>'transactionId' AS xid,
+ a->>'requestType' AS operation,
+ a->>'user' AS usename,
+ a->>'serverName' AS server,
+ a->>'profile' AS profile,
+ a->>'schemaName' AS schema_name,
+ a->>'tableName' AS table_name,
+ a->>'dataSource' AS data_source,
+ to_timestamp((a->>'startTimeMs')::bigint / 1000.0) AS query_start,
+ a->>'host' AS pxf_host
+FROM (
+ SELECT json_array_elements(raw::json -> 'activities') AS a
+ FROM pxf_stat_activity_raw() AS raw
+) s;
+
+-- Per-segment cancellation primitives backing pxf_cancel_backend /
+-- pxf_interrupt_backend. Each segment asks its local PXF instance to terminate
+-- the in-flight requests of the given Cloudberry session that originate from
its
+-- own segment id, returning the JSON body verbatim as a single row (e.g.
+-- {"cancelled":N} / {"interrupted":N}). Dispatched to every segment; the
counts
+-- are summed by the SQL wrappers below. Set-returning because EXECUTE ON ALL
+-- SEGMENTS is only permitted for set-returning functions.
+CREATE FUNCTION pxf_cancel_backend_raw(session_id int) RETURNS SETOF text
+AS 'MODULE_PATHNAME', 'pxf_cancel_backend_raw'
+LANGUAGE C VOLATILE STRICT EXECUTE ON ALL SEGMENTS;
+
+CREATE FUNCTION pxf_interrupt_backend_raw(session_id int) RETURNS SETOF text
+AS 'MODULE_PATHNAME', 'pxf_interrupt_backend_raw'
+LANGUAGE C VOLATILE STRICT EXECUTE ON ALL SEGMENTS;
+
+-- Gracefully cancels the in-flight PXF requests of a Cloudberry session across
+-- the whole cluster by ending their current bridge. Returns the number of
+-- requests that were signalled. Analogous to pg_cancel_backend, but keyed by
+-- the Cloudberry session id (as reported in pxf_stat_activity.session_id).
+CREATE FUNCTION pxf_cancel_backend(session_id int) RETURNS int AS $$
+ SELECT coalesce(sum((raw::json ->> 'cancelled')::int), 0)::int
+ FROM pxf_cancel_backend_raw(session_id) AS raw
+$$ LANGUAGE sql VOLATILE;
+
+-- Interrupts the worker thread(s) of the in-flight PXF requests of a
Cloudberry
+-- session across the whole cluster. Returns the number of requests that were
+-- interrupted. A forceful complement to pxf_cancel_backend for requests that
do
+-- not observe cancellation (e.g. blocked in a non-interruptible read).
+CREATE FUNCTION pxf_interrupt_backend(session_id int) RETURNS int AS $$
+ SELECT coalesce(sum((raw::json ->> 'interrupted')::int), 0)::int
+ FROM pxf_interrupt_backend_raw(session_id) AS raw
+$$ LANGUAGE sql VOLATILE;
+
+-- Functions default to EXECUTE for PUBLIC, which would let any user observe
and
+-- cancel other users' PXF requests (session ids are trivially enumerable).
+-- Superusers bypass ACLs; delegate explicitly with e.g.
+-- GRANT EXECUTE ON FUNCTION pxf_cancel_backend(int) TO monitoring_role;
+-- The wrappers are plain SQL, so the _raw primitives must be revoked as well.
+REVOKE ALL ON FUNCTION pxf_stat_activity_raw() FROM PUBLIC;
+REVOKE ALL ON FUNCTION pxf_cancel_backend_raw(int) FROM PUBLIC;
+REVOKE ALL ON FUNCTION pxf_interrupt_backend_raw(int) FROM PUBLIC;
+REVOKE ALL ON FUNCTION pxf_cancel_backend(int) FROM PUBLIC;
+REVOKE ALL ON FUNCTION pxf_interrupt_backend(int) FROM PUBLIC;
diff --git a/external-table/pxf--2.2--2.1.sql b/external-table/pxf--2.2--2.1.sql
new file mode 100644
index 00000000..5bcacf59
--- /dev/null
+++ b/external-table/pxf--2.2--2.1.sql
@@ -0,0 +1,39 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+/* external_table/pxf--2.2--2.1.sql */
+
+------------------------------------------------------------------
+-- PXF Activity Monitoring
+------------------------------------------------------------------
+
+-- remove the objects from the extension
+ALTER EXTENSION pxf DROP FUNCTION pxf_interrupt_backend(int);
+ALTER EXTENSION pxf DROP FUNCTION pxf_cancel_backend(int);
+ALTER EXTENSION pxf DROP FUNCTION pxf_interrupt_backend_raw(int);
+ALTER EXTENSION pxf DROP FUNCTION pxf_cancel_backend_raw(int);
+ALTER EXTENSION pxf DROP VIEW pxf_stat_activity;
+ALTER EXTENSION pxf DROP FUNCTION pxf_stat_activity_raw();
+
+-- remove the objects themselves from the catalog; the view depends on
+-- pxf_stat_activity_raw(), so it has to go first
+DROP FUNCTION pxf_interrupt_backend(int);
+DROP FUNCTION pxf_cancel_backend(int);
+DROP FUNCTION pxf_interrupt_backend_raw(int);
+DROP FUNCTION pxf_cancel_backend_raw(int);
+DROP VIEW pxf_stat_activity;
+DROP FUNCTION pxf_stat_activity_raw();
diff --git a/external-table/pxf--2.2.sql b/external-table/pxf--2.2.sql
new file mode 100644
index 00000000..927f42bf
--- /dev/null
+++ b/external-table/pxf--2.2.sql
@@ -0,0 +1,130 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+------------------------------------------------------------------
+-- PXF Protocol/Formatters
+------------------------------------------------------------------
+
+CREATE OR REPLACE FUNCTION pg_catalog.pxf_write() RETURNS integer
+AS 'MODULE_PATHNAME', 'pxfprotocol_export'
+LANGUAGE C STABLE;
+
+CREATE OR REPLACE FUNCTION pg_catalog.pxf_read() RETURNS integer
+AS 'MODULE_PATHNAME', 'pxfprotocol_import'
+LANGUAGE C STABLE;
+
+CREATE OR REPLACE FUNCTION pg_catalog.pxf_validate() RETURNS void
+AS 'MODULE_PATHNAME', 'pxfprotocol_validate_urls'
+LANGUAGE C STABLE;
+
+CREATE OR REPLACE FUNCTION pg_catalog.pxfwritable_import() RETURNS record
+AS 'MODULE_PATHNAME', 'gpdbwritableformatter_import'
+LANGUAGE C STABLE;
+
+CREATE OR REPLACE FUNCTION pg_catalog.pxfwritable_export(record) RETURNS bytea
+AS 'MODULE_PATHNAME', 'gpdbwritableformatter_export'
+LANGUAGE C STABLE;
+
+CREATE OR REPLACE FUNCTION pg_catalog.pxfdelimited_import() RETURNS record
+AS 'MODULE_PATHNAME', 'pxfdelimited_import'
+LANGUAGE C STABLE;
+
+CREATE TRUSTED PROTOCOL pxf (
+ writefunc = pxf_write,
+ readfunc = pxf_read,
+ validatorfunc = pxf_validate);
+
+------------------------------------------------------------------
+-- PXF Activity Monitoring
+------------------------------------------------------------------
+
+-- Raw per-segment accessor: each segment asks its local PXF instance for the
+-- activity that originates from its own segment id and returns the JSON body
+-- verbatim as a single row. Dispatched to every segment; the typed columns are
+-- produced by the pxf_stat_activity view below. The function is set-returning
+-- (one row per segment) because EXECUTE ON ALL SEGMENTS is only permitted for
+-- set-returning functions.
+CREATE FUNCTION pxf_stat_activity_raw() RETURNS SETOF text
+AS 'MODULE_PATHNAME', 'pxf_stat_activity_raw'
+LANGUAGE C VOLATILE EXECUTE ON ALL SEGMENTS;
+
+-- pg_stat_activity-like view of the queries currently running inside PXF,
+-- aggregated across all segment hosts. DISTINCT is a safety net; PXF already
+-- de-duplicates by filtering each response to the requesting segment id.
+CREATE VIEW pxf_stat_activity AS
+SELECT DISTINCT
+ (a->>'segmentId')::int AS segment_id,
+ (a->>'gpSessionId')::int AS session_id,
+ (a->>'gpCommandCount')::int AS command_count,
+ a->>'transactionId' AS xid,
+ a->>'requestType' AS operation,
+ a->>'user' AS usename,
+ a->>'serverName' AS server,
+ a->>'profile' AS profile,
+ a->>'schemaName' AS schema_name,
+ a->>'tableName' AS table_name,
+ a->>'dataSource' AS data_source,
+ to_timestamp((a->>'startTimeMs')::bigint / 1000.0) AS query_start,
+ a->>'host' AS pxf_host
+FROM (
+ SELECT json_array_elements(raw::json -> 'activities') AS a
+ FROM pxf_stat_activity_raw() AS raw
+) s;
+
+-- Per-segment cancellation primitives backing pxf_cancel_backend /
+-- pxf_interrupt_backend. Each segment asks its local PXF instance to terminate
+-- the in-flight requests of the given Cloudberry session that originate from
its
+-- own segment id, returning the JSON body verbatim as a single row (e.g.
+-- {"cancelled":N} / {"interrupted":N}). Dispatched to every segment; the
counts
+-- are summed by the SQL wrappers below. Set-returning because EXECUTE ON ALL
+-- SEGMENTS is only permitted for set-returning functions.
+CREATE FUNCTION pxf_cancel_backend_raw(session_id int) RETURNS SETOF text
+AS 'MODULE_PATHNAME', 'pxf_cancel_backend_raw'
+LANGUAGE C VOLATILE STRICT EXECUTE ON ALL SEGMENTS;
+
+CREATE FUNCTION pxf_interrupt_backend_raw(session_id int) RETURNS SETOF text
+AS 'MODULE_PATHNAME', 'pxf_interrupt_backend_raw'
+LANGUAGE C VOLATILE STRICT EXECUTE ON ALL SEGMENTS;
+
+-- Gracefully cancels the in-flight PXF requests of a Cloudberry session across
+-- the whole cluster by ending their current bridge. Returns the number of
+-- requests that were signalled. Analogous to pg_cancel_backend, but keyed by
+-- the Cloudberry session id (as reported in pxf_stat_activity.session_id).
+CREATE FUNCTION pxf_cancel_backend(session_id int) RETURNS int AS $$
+ SELECT coalesce(sum((raw::json ->> 'cancelled')::int), 0)::int
+ FROM pxf_cancel_backend_raw(session_id) AS raw
+$$ LANGUAGE sql VOLATILE;
+
+-- Interrupts the worker thread(s) of the in-flight PXF requests of a
Cloudberry
+-- session across the whole cluster. Returns the number of requests that were
+-- interrupted. A forceful complement to pxf_cancel_backend for requests that
do
+-- not observe cancellation (e.g. blocked in a non-interruptible read).
+CREATE FUNCTION pxf_interrupt_backend(session_id int) RETURNS int AS $$
+ SELECT coalesce(sum((raw::json ->> 'interrupted')::int), 0)::int
+ FROM pxf_interrupt_backend_raw(session_id) AS raw
+$$ LANGUAGE sql VOLATILE;
+
+-- Functions default to EXECUTE for PUBLIC, which would let any user observe
and
+-- cancel other users' PXF requests (session ids are trivially enumerable).
+-- Superusers bypass ACLs; delegate explicitly with e.g.
+-- GRANT EXECUTE ON FUNCTION pxf_cancel_backend(int) TO monitoring_role;
+-- The wrappers are plain SQL, so the _raw primitives must be revoked as well.
+REVOKE ALL ON FUNCTION pxf_stat_activity_raw() FROM PUBLIC;
+REVOKE ALL ON FUNCTION pxf_cancel_backend_raw(int) FROM PUBLIC;
+REVOKE ALL ON FUNCTION pxf_interrupt_backend_raw(int) FROM PUBLIC;
+REVOKE ALL ON FUNCTION pxf_cancel_backend(int) FROM PUBLIC;
+REVOKE ALL ON FUNCTION pxf_interrupt_backend(int) FROM PUBLIC;
diff --git a/external-table/pxf.control b/external-table/pxf.control
index ca0ae18a..ef773738 100644
--- a/external-table/pxf.control
+++ b/external-table/pxf.control
@@ -1,5 +1,5 @@
directory = 'extension'
-default_version = '2.1'
+default_version = '2.2'
comment = 'Extension which allows to access unmanaged data'
module_pathname = '$libdir/pxf'
superuser = true
diff --git a/external-table/src/pxf_cancel_activity.c
b/external-table/src/pxf_cancel_activity.c
new file mode 100644
index 00000000..4bb16abe
--- /dev/null
+++ b/external-table/src/pxf_cancel_activity.c
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/*
+ * pxf_cancel_activity.c
+ *
+ * Backs the pxf_cancel_backend and pxf_interrupt_backend SQL functions. Each
is
+ * dispatched with EXECUTE ON ALL SEGMENTS: every segment asks its local PXF
+ * instance to terminate the in-flight requests of a given Greenplum session
+ * that originate from its own segment id (passed via X-GP-SEGMENT-ID).
Because a
+ * single PXF instance serves every segment co-located on the host, this
+ * per-segment filter guarantees each running request is acted on exactly once,
+ * by its owning segment.
+ *
+ * Both functions are set-returning (one row per segment, carrying the PXF
+ * response verbatim as a JSON text value) because EXECUTE ON ALL SEGMENTS is
+ * only permitted for set-returning functions; the counts are summed by the SQL
+ * wrappers, keeping this layer schema-agnostic.
+ */
+
+#include "libchurl.h"
+#include "pxfuriparser.h"
+#include "pxfutils.h"
+
+#include "funcapi.h"
+#include "cdb/cdbvars.h"
+#include "utils/builtins.h"
+
+#define PXF_CANCEL_READ_BUFFER_SIZE (64 * 1024)
+
+PG_FUNCTION_INFO_V1(pxf_cancel_backend_raw);
+PG_FUNCTION_INFO_V1(pxf_interrupt_backend_raw);
+
+Datum pxf_cancel_backend_raw(PG_FUNCTION_ARGS);
+Datum pxf_interrupt_backend_raw(PG_FUNCTION_ARGS);
+
+/*
+ * Issues an HTTP GET to the given local PXF endpoint, scoped to this segment
and
+ * the target Greenplum session, and returns the raw JSON response body as a
+ * palloc'd text value in the current memory context.
+ */
+static text *
+fetch_backend_control_body(const char *endpoint, int32 session_id)
+{
+ CHURL_HEADERS headers;
+ CHURL_HANDLE handle;
+ StringInfoData uri;
+ StringInfoData response;
+ char readbuf[PXF_CANCEL_READ_BUFFER_SIZE];
+ char segment_id[32];
+ char session_buf[32];
+ size_t n;
+ text *result;
+
+ /* build the request URI for the local PXF instance */
+ initStringInfo(&uri);
+ appendStringInfo(&uri, "http://%s/%s/%s";,
+ get_authority(), PXF_SERVICE_PREFIX,
endpoint);
+
+ /* scope the request to this segment so co-located segments don't act
twice */
+ snprintf(segment_id, sizeof(segment_id), "%d", GpIdentity.segindex);
+ snprintf(session_buf, sizeof(session_buf), "%d", session_id);
+
+ headers = churl_headers_init();
+ churl_headers_append(headers, "X-GP-SEGMENT-ID", segment_id);
+ churl_headers_append(headers, "X-GP-SESSION-ID", session_buf);
+ churl_headers_append(headers, "Accept", "application/json");
+
+ elog(DEBUG2, "%s: segment %d requesting %s for session %d",
+ endpoint, GpIdentity.segindex, uri.data, session_id);
+
+ handle = churl_init_download(uri.data, headers);
+
+ /* read some bytes to make sure the connection is established */
+ churl_read_check_connectivity(handle);
+
+ /* read the full JSON body */
+ initStringInfo(&response);
+ while ((n = churl_read(handle, readbuf, sizeof(readbuf))) != 0)
+ appendBinaryStringInfo(&response, readbuf, n);
+
+ /* surface any error reported by the PXF service on the closed
connection */
+ churl_read_check_connectivity(handle);
+
+ churl_cleanup(handle, false);
+ churl_headers_cleanup(headers);
+
+ result = cstring_to_text(response.data);
+
+ pfree(uri.data);
+ pfree(response.data);
+
+ return result;
+}
+
+/*
+ * Shared set-returning body: emit a single row carrying the raw JSON response
+ * from the local PXF instance for the given endpoint and session id.
Dispatched
+ * with EXECUTE ON ALL SEGMENTS (which Cloudberry only permits for
set-returning
+ * functions); the pxf_cancel_backend / pxf_interrupt_backend SQL wrappers sum
+ * the per-segment counts.
+ */
+static Datum
+backend_control_srf(FunctionCallInfo fcinfo, const char *endpoint)
+{
+ FuncCallContext *funcctx;
+
+ if (SRF_IS_FIRSTCALL())
+ {
+ MemoryContext oldcontext;
+ int32 session_id = PG_GETARG_INT32(0);
+
+ funcctx = SRF_FIRSTCALL_INIT();
+
+ /* the fetched body must outlive this call, so build it in the
+ * multi-call context and hand it back one row at a time */
+ oldcontext =
MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+ funcctx->user_fctx = fetch_backend_control_body(endpoint,
session_id);
+ funcctx->max_calls = 1;
+
+ MemoryContextSwitchTo(oldcontext);
+ }
+
+ funcctx = SRF_PERCALL_SETUP();
+
+ if (funcctx->call_cntr < funcctx->max_calls)
+ SRF_RETURN_NEXT(funcctx, PointerGetDatum((text *)
funcctx->user_fctx));
+
+ SRF_RETURN_DONE(funcctx);
+}
+
+/*
+ * pxf_cancel_backend_raw(session_id int)
+ *
+ * Asks the local PXF instance to gracefully cancel the in-flight requests of
the
+ * given session (by ending their current bridge). Returns the raw JSON body,
+ * e.g. {"cancelled":N}.
+ */
+Datum
+pxf_cancel_backend_raw(PG_FUNCTION_ARGS)
+{
+ return backend_control_srf(fcinfo, "cancel_backend");
+}
+
+/*
+ * pxf_interrupt_backend_raw(session_id int)
+ *
+ * Asks the local PXF instance to interrupt the worker thread(s) of the
in-flight
+ * requests of the given session. Returns the raw JSON body, e.g.
+ * {"interrupted":N}.
+ */
+Datum
+pxf_interrupt_backend_raw(PG_FUNCTION_ARGS)
+{
+ return backend_control_srf(fcinfo, "interrupt_backend");
+}
diff --git a/external-table/src/pxf_stat_activity.c
b/external-table/src/pxf_stat_activity.c
new file mode 100644
index 00000000..21ca7620
--- /dev/null
+++ b/external-table/src/pxf_stat_activity.c
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/*
+ * pxf_stat_activity.c
+ *
+ * Backs the pxf_stat_activity SQL view. The function pxf_stat_activity_raw()
+ * is dispatched with EXECUTE ON ALL SEGMENTS: each segment asks its local PXF
+ * instance for the requests that originate from its own segment id (passed via
+ * the X-GP-SEGMENT-ID header). Because a single PXF instance serves every
+ * segment co-located on the host, this per-segment filter guarantees each
+ * running query is reported exactly once, by its owning segment.
+ *
+ * The function is set-returning (one row per segment, carrying the PXF
response
+ * verbatim as a JSON text value) because EXECUTE ON ALL SEGMENTS is only
+ * permitted for set-returning functions; the typed columns are produced by the
+ * SQL view, keeping this layer schema-agnostic.
+ */
+
+#include "libchurl.h"
+#include "pxfuriparser.h"
+#include "pxfutils.h"
+
+#include "funcapi.h"
+#include "cdb/cdbvars.h"
+#include "utils/builtins.h"
+
+#define PXF_STAT_READ_BUFFER_SIZE (64 * 1024)
+
+PG_FUNCTION_INFO_V1(pxf_stat_activity_raw);
+
+Datum pxf_stat_activity_raw(PG_FUNCTION_ARGS);
+
+/*
+ * Issues an HTTP GET to the local PXF /pxf/stat_activity endpoint, scoped to
+ * this segment, and returns the raw JSON response body as a palloc'd text
value
+ * in the current memory context.
+ */
+static text *
+fetch_stat_activity_body(void)
+{
+ CHURL_HEADERS headers;
+ CHURL_HANDLE handle;
+ StringInfoData uri;
+ StringInfoData response;
+ char readbuf[PXF_STAT_READ_BUFFER_SIZE];
+ char segment_id[32];
+ size_t n;
+ text *result;
+
+ /* build the request URI for the local PXF instance */
+ initStringInfo(&uri);
+ appendStringInfo(&uri, "http://%s/%s/stat_activity";,
+ get_authority(), PXF_SERVICE_PREFIX);
+
+ /* scope the request to this segment so co-located segments don't
duplicate */
+ snprintf(segment_id, sizeof(segment_id), "%d", GpIdentity.segindex);
+
+ headers = churl_headers_init();
+ churl_headers_append(headers, "X-GP-SEGMENT-ID", segment_id);
+ churl_headers_append(headers, "Accept", "application/json");
+
+ elog(DEBUG2, "pxf_stat_activity: segment %d requesting %s",
+ GpIdentity.segindex, uri.data);
+
+ handle = churl_init_download(uri.data, headers);
+
+ /* read some bytes to make sure the connection is established */
+ churl_read_check_connectivity(handle);
+
+ /* read the full JSON body */
+ initStringInfo(&response);
+ while ((n = churl_read(handle, readbuf, sizeof(readbuf))) != 0)
+ appendBinaryStringInfo(&response, readbuf, n);
+
+ /* surface any error reported by the PXF service on the closed
connection */
+ churl_read_check_connectivity(handle);
+
+ churl_cleanup(handle, false);
+ churl_headers_cleanup(headers);
+
+ result = cstring_to_text(response.data);
+
+ pfree(uri.data);
+ pfree(response.data);
+
+ return result;
+}
+
+/*
+ * pxf_stat_activity_raw
+ *
+ * Set-returning function dispatched with EXECUTE ON ALL SEGMENTS (which
+ * Cloudberry only permits for set-returning functions). Each segment emits a
+ * single row: the raw JSON response body from its local PXF instance, scoped
to
+ * that segment. The pxf_stat_activity view unions these rows and expands them
+ * into typed columns.
+ */
+Datum
+pxf_stat_activity_raw(PG_FUNCTION_ARGS)
+{
+ FuncCallContext *funcctx;
+
+ if (SRF_IS_FIRSTCALL())
+ {
+ MemoryContext oldcontext;
+
+ funcctx = SRF_FIRSTCALL_INIT();
+
+ /* the fetched body must outlive this call, so build it in the
+ * multi-call context and hand it back one row at a time */
+ oldcontext =
MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+ funcctx->user_fctx = fetch_stat_activity_body();
+ funcctx->max_calls = 1;
+
+ MemoryContextSwitchTo(oldcontext);
+ }
+
+ funcctx = SRF_PERCALL_SETUP();
+
+ if (funcctx->call_cntr < funcctx->max_calls)
+ SRF_RETURN_NEXT(funcctx, PointerGetDatum((text *)
funcctx->user_fctx));
+
+ SRF_RETURN_DONE(funcctx);
+}
diff --git
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/activity/ActiveRequest.java
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/activity/ActiveRequest.java
new file mode 100644
index 00000000..c1d0bf40
--- /dev/null
+++
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/activity/ActiveRequest.java
@@ -0,0 +1,167 @@
+package org.apache.cloudberry.pxf.service.activity;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.cloudberry.pxf.service.bridge.Bridge;
+
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * A single in-flight PXF request tracked by {@link ActiveRequestRegistry}.
+ * <p>
+ * In addition to the immutable {@link ActiveRequestInfo} snapshot (which
defines
+ * the `pxf_stat_activity` JSON wire contract) it holds the two live
+ * handles used to terminate the request from another thread:
+ * - the worker `thread` that is processing it (for `pxf_interrupt_backend`)
+ * - the current `bridge` (for `pxf_cancel_backend`)
+ */
+@Slf4j
+class ActiveRequest {
+
+ /*
+ * Concurrency: PXF processes a request synchronously on a servlet
+ * thread that is returned to a pool afterwards, so a naive
+ * `thread.interrupt()` could land on an unrelated request the thread has
+ * since picked up. To prevent that, every mutation that races with request
+ * completion is serialized on `lock` and gated by the `finished`
+ * flag:
+ * <ul>
+ * <li>the worker calls `markFinished()` in its `finally` block
+ * before it can return to the pool;</li>
+ * <li>`cancelIfActive()` / `interruptIfActive()` only act while
+ * `finished == false`, which — because both sides take the same
+ * lock — guarantees the worker is still executing this request.</li>
+ * </ul>
+ */
+
+ /** Immutable snapshot exposed via {@code pxf_stat_activity}. */
+ final ActiveRequestInfo info;
+
+ /** The servlet worker thread that is processing this request. */
+ private final Thread thread;
+
+ /** Serializes cancellation/interruption against request completion. */
+ private final ReentrantLock lock = new ReentrantLock();
+
+ /** Set once the request has completed; guarded by {@link #lock}. */
+ private boolean finished;
+
+ /** Set when the request has been asked to stop; read by the worker loop.
*/
+ private volatile boolean cancelled;
+
+ /** The bridge currently in use by the worker; guarded by {@link #lock}. */
+ private Bridge bridge;
+
+ ActiveRequest(ActiveRequestInfo info, Thread thread) {
+ this.info = info;
+ this.thread = thread;
+ }
+
+ /**
+ * @return whether this request has been asked to stop; polled by the
worker
+ * loop so it can break out between fragments/records without waiting for a
+ * blocking read to be unblocked by {@link #cancelIfActive()}.
+ */
+ boolean isCancelled() {
+ return cancelled;
+ }
+
+ /**
+ * Records (or clears, with {@code null}) the bridge the worker is
currently
+ * iterating over, so that {@link #cancelIfActive()} can end it.
+ */
+ void setBridge(Bridge bridge) {
+ lock.lock();
+ try {
+ this.bridge = bridge;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Marks the request as completed. After this returns, {@link
#cancelIfActive()}
+ * and {@link #interruptIfActive()} become no-ops, so the worker thread
can be
+ * safely returned to the pool without risking a stray interrupt.
+ */
+ void markFinished() {
+ lock.lock();
+ try {
+ finished = true;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Requests cancellation of this request if it is still active: raises the
+ * {@link #cancelled} flag and ends the current bridge to unblock a pending
+ * read. Safe to call concurrently with the worker's own bridge lifecycle;
+ * {@code endIteration()} must therefore be idempotent enough to tolerate a
+ * double close (its error is logged and swallowed).
+ *
+ * @return {@code true} if the request was still active and was signalled
+ */
+ boolean cancelIfActive() {
+ lock.lock();
+ try {
+ if (finished) {
+ return false;
+ }
+ cancelled = true;
+ if (bridge != null) {
+ try {
+ bridge.endIteration();
+ } catch (Exception e) {
+ log.warn("Ignoring error while cancelling bridge for {}",
describe(), e);
+ }
+ }
+ return true;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Interrupts the worker thread if the request is still active. Holding the
+ * lock while checking {@link #finished} guarantees the thread has not yet
+ * returned to the pool, so the interrupt cannot leak onto a later request.
+ *
+ * @return {@code true} if the request was still active and was interrupted
+ */
+ boolean interruptIfActive() {
+ lock.lock();
+ try {
+ if (finished) {
+ return false;
+ }
+ thread.interrupt();
+ return true;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private String describe() {
+ return String.format("session %d, segment %d, xid %s",
+ info.getGpSessionId(), info.getSegmentId(),
info.getTransactionId());
+ }
+}
diff --git
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/activity/ActiveRequestInfo.java
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/activity/ActiveRequestInfo.java
new file mode 100644
index 00000000..38fdcfc2
--- /dev/null
+++
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/activity/ActiveRequestInfo.java
@@ -0,0 +1,96 @@
+package org.apache.cloudberry.pxf.service.activity;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import lombok.Getter;
+import org.apache.cloudberry.pxf.api.model.RequestContext;
+
+/**
+ * Immutable snapshot of an in-flight PXF request, captured at the moment the
+ * request started being processed. Instances are exposed via the
+ * `/pxf/stat_activity` endpoint.
+ * <p>
+ * The getters are serialized to JSON by Jackson.
+ */
+@Getter
+public class ActiveRequestInfo {
+
+ /** The Cloudberry segment id that originated the request. */
+ private final int segmentId;
+
+ /** The Cloudberry session id (aka ssid). */
+ private final int gpSessionId;
+
+ /** The Cloudberry command count within the session (aka ccnt). */
+ private final int gpCommandCount;
+
+ /** The Cloudberry transaction id (XID) of the originating query. */
+ private final String transactionId;
+
+ /** The kind of operation: READ_BRIDGE or WRITE_BRIDGE. */
+ private final String requestType;
+
+ /** The end-user identity that issued the request. */
+ private final String user;
+
+ /** The name of the PXF server configuration used by the request. */
+ private final String serverName;
+
+ /** The profile associated with the request (e.g. hdfs:text). */
+ private final String profile;
+
+ /** The originating Cloudberry schema name. */
+ private final String schemaName;
+
+ /** The originating Cloudberry table name. */
+ private final String tableName;
+
+ /** The data source (file path or external resource identifier). */
+ private final String dataSource;
+
+ /** Epoch milliseconds when the request started being processed. */
+ private final long startTimeMs;
+
+ /** The hostname of the PXF instance that is serving the request. */
+ private final String host;
+
+ /**
+ * Captures a snapshot of the relevant fields from the request context.
+ *
+ * @param context the request context of the in-flight request
+ * @param startTimeMs epoch milliseconds when processing started
+ * @param host hostname of the PXF instance serving the request
+ */
+ public ActiveRequestInfo(RequestContext context, long startTimeMs, String
host) {
+ this.segmentId = context.getSegmentId();
+ this.gpSessionId = context.getGpSessionId();
+ this.gpCommandCount = context.getGpCommandCount();
+ this.transactionId = context.getTransactionId();
+ this.requestType = context.getRequestType() == null ? null :
context.getRequestType().name();
+ this.user = context.getUser();
+ this.serverName = context.getServerName();
+ this.profile = context.getProfile();
+ this.schemaName = context.getSchemaName();
+ this.tableName = context.getTableName();
+ this.dataSource = context.getDataSource();
+ this.startTimeMs = startTimeMs;
+ this.host = host;
+ }
+}
diff --git
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/activity/ActiveRequestRegistry.java
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/activity/ActiveRequestRegistry.java
new file mode 100644
index 00000000..22f7d2f7
--- /dev/null
+++
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/activity/ActiveRequestRegistry.java
@@ -0,0 +1,191 @@
+package org.apache.cloudberry.pxf.service.activity;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.cloudberry.pxf.api.model.RequestContext;
+import org.apache.cloudberry.pxf.service.bridge.Bridge;
+import org.springframework.stereotype.Component;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Registry of the requests that are currently being processed by this PXF
instance.
+ * A request is registered when it starts being processed by
+ * `BaseServiceImpl.processData()` and unregistered when it ends.
+ */
+@Component
+@Slf4j
+public class ActiveRequestRegistry {
+
+ /** Sentinel meaning "all segments" for the segment-id filter. */
+ public static final int ALL_SEGMENTS = -1;
+
+ /** All in-flight requests on this instance. */
+ private final Set<ActiveRequest> activeRequests =
ConcurrentHashMap.newKeySet();
+
+ /**
+ * The request currently being processed by the calling worker thread, used
+ * by the worker itself `attachBridge`, `isCurrentCancelled`) and to locate
+ * the entry to remove on `unregister` without threading a handle through
the
+ * processing call chain. Set on `register` and cleared on `unregister`.
+ */
+ private final ThreadLocal<ActiveRequest> currentRequest = new
ThreadLocal<>();
+
+ private final String hostname;
+
+ public ActiveRequestRegistry() {
+ this.hostname = resolveHostname();
+ }
+
+ /**
+ * Registers the request being processed by the calling thread as
in-flight.
+ * Must be paired with a call to `unregister()` in a finally block, on the
+ * same thread.
+ *
+ * @param context the request context of the request being processed
+ */
+ public void register(RequestContext context) {
+ ActiveRequestInfo info = new ActiveRequestInfo(context,
System.currentTimeMillis(), hostname);
+ ActiveRequest request = new ActiveRequest(info,
Thread.currentThread());
+ activeRequests.add(request);
+ currentRequest.set(request);
+ }
+
+ /**
+ * Removes the request registered by the calling thread from the registry
and
+ * marks it finished so that any concurrent `cancel()` / `interrupt` can no
+ * longer act on the (soon to be recycled) worker thread. No-op if the
calling
+ * thread has no registered request.
+ */
+ public void unregister() {
+ ActiveRequest request = currentRequest.get();
+ if (request != null) {
+ activeRequests.remove(request);
+ request.markFinished();
+ }
+ currentRequest.remove();
+ }
+
+ /**
+ * Records the bridge the calling worker thread is currently iterating
over,
+ * so that a concurrent `cancel` can end it. Pass `null` to clear
+ * the reference once the bridge is closed. No-op if the caller is not a
+ * registered worker thread.
+ *
+ * @param bridge the bridge in use, or `null` to detach
+ */
+ public void attachBridge(Bridge bridge) {
+ ActiveRequest request = currentRequest.get();
+ if (request != null) {
+ request.setBridge(bridge);
+ }
+ }
+
+ /**
+ * @return whether the request being processed by the calling worker thread
+ * has been asked to cancel. Polled by the read/write loops so they stop
+ * between fragments/records.
+ */
+ public boolean isCurrentCancelled() {
+ ActiveRequest request = currentRequest.get();
+ return request != null && request.isCancelled();
+ }
+
+ /**
+ * Returns a snapshot of the currently active requests, optionally filtered
+ * by the originating segment id.
+ *
+ * @param segmentId the originating segment id to filter by, or
+ * `ALL_SEGMENTS` to return activity for all segments
+ * @return list of active request descriptors
+ */
+ public List<ActiveRequestInfo> snapshot(int segmentId) {
+ List<ActiveRequestInfo> result = new
ArrayList<>(activeRequests.size());
+ for (ActiveRequest request : activeRequests) {
+ if (matchesSegment(request, segmentId)) {
+ result.add(request.info);
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Gracefully cancels every active request of the given Cloudberry session
on
+ * this segment by ending its current bridge (see
`ActiveRequest.cancelIfActive()`).
+ *
+ * @param segmentId the originating segment id, or `ALL_SEGMENTS`
+ * @param sessionId the Cloudberry session id whose requests to cancel
+ * @return the number of active requests that were signalled
+ */
+ public int cancel(int segmentId, int sessionId) {
+ int count = 0;
+ for (ActiveRequest request : activeRequests) {
+ if (matchesSegment(request, segmentId) &&
request.info.getGpSessionId() == sessionId) {
+ if (request.cancelIfActive()) {
+ count++;
+ }
+ }
+ }
+ log.info("pxf_cancel_backend: signalled {} request(s) for session {}
on segment {}",
+ count, sessionId, segmentId);
+ return count;
+ }
+
+ /**
+ * Interrupts the worker thread of every active request of the given
+ * Cloudberry session on this segment (see
`ActiveRequest.interruptIfActive()`).
+ *
+ * @param segmentId the originating segment id, or `ALL_SEGMENTS`
+ * @param sessionId the Cloudberry session id whose requests to interrupt
+ * @return the number of active requests that were interrupted
+ */
+ public int interrupt(int segmentId, int sessionId) {
+ int count = 0;
+ for (ActiveRequest request : activeRequests) {
+ if (matchesSegment(request, segmentId) &&
request.info.getGpSessionId() == sessionId) {
+ if (request.interruptIfActive()) {
+ count++;
+ }
+ }
+ }
+ log.info("pxf_interrupt_backend: interrupted {} request(s) for session
{} on segment {}",
+ count, sessionId, segmentId);
+ return count;
+ }
+
+ private static boolean matchesSegment(ActiveRequest request, int
segmentId) {
+ return segmentId == ALL_SEGMENTS || request.info.getSegmentId() ==
segmentId;
+ }
+
+ private static String resolveHostname() {
+ try {
+ return InetAddress.getLocalHost().getHostName();
+ } catch (UnknownHostException e) {
+ log.warn("Unable to resolve local hostname for pxf_stat_activity
reporting", e);
+ return "unknown";
+ }
+ }
+}
diff --git
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/BaseServiceImpl.java
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/BaseServiceImpl.java
index fb61d91b..fdb9c0f6 100644
---
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/BaseServiceImpl.java
+++
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/BaseServiceImpl.java
@@ -6,6 +6,7 @@ import org.apache.cloudberry.pxf.api.model.ConfigurationFactory;
import org.apache.cloudberry.pxf.api.model.RequestContext;
import org.apache.cloudberry.pxf.api.utilities.Utilities;
import org.apache.cloudberry.pxf.service.MetricsReporter;
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestRegistry;
import org.apache.cloudberry.pxf.service.bridge.Bridge;
import org.apache.cloudberry.pxf.service.bridge.BridgeFactory;
import org.apache.cloudberry.pxf.service.security.SecurityService;
@@ -26,26 +27,30 @@ public abstract class BaseServiceImpl<T> extends
PxfErrorReporter<T> {
private final ConfigurationFactory configurationFactory;
private final BridgeFactory bridgeFactory;
private final SecurityService securityService;
+ private final ActiveRequestRegistry activeRequestRegistry;
/**
* Creates a new instance of the service with auto-wired dependencies.
*
- * @param serviceName name of the service
- * @param configurationFactory configuration factory
- * @param bridgeFactory bridge factory
- * @param securityService security service
- * @param metricsReporter metrics reporter service
+ * @param serviceName name of the service
+ * @param configurationFactory configuration factory
+ * @param bridgeFactory bridge factory
+ * @param securityService security service
+ * @param metricsReporter metrics reporter service
+ * @param activeRequestRegistry registry of in-flight requests
*/
protected BaseServiceImpl(String serviceName,
ConfigurationFactory configurationFactory,
BridgeFactory bridgeFactory,
SecurityService securityService,
- MetricsReporter metricsReporter) {
+ MetricsReporter metricsReporter,
+ ActiveRequestRegistry activeRequestRegistry) {
this.serviceName = serviceName;
this.configurationFactory = configurationFactory;
this.bridgeFactory = bridgeFactory;
this.securityService = securityService;
this.metricsReporter = metricsReporter;
+ this.activeRequestRegistry = activeRequestRegistry;
}
/**
@@ -70,8 +75,19 @@ public abstract class BaseServiceImpl<T> extends
PxfErrorReporter<T> {
Instant startTime = Instant.now();
- // execute processing action with a proper identity
- OperationResult result = securityService.doAs(context, action);
+ // clear any interrupt status left on this pooled worker thread by a
+ // previous request that was targeted by pxf_interrupt_backend, so it
+ // cannot leak into the request we are about to process
+ Thread.interrupted();
+
+ activeRequestRegistry.register(context);
+ OperationResult result;
+ try {
+ // execute processing action with a proper identity
+ result = securityService.doAs(context, action);
+ } finally {
+ activeRequestRegistry.unregister();
+ }
// obtain results after executing the action
OperationStats stats = result.getStats();
@@ -115,4 +131,24 @@ public abstract class BaseServiceImpl<T> extends
PxfErrorReporter<T> {
protected Bridge getBridge(RequestContext context) {
return bridgeFactory.getBridge(context);
}
+
+ /**
+ * Records the bridge the current worker thread is iterating over so that a
+ * concurrent `pxf_cancel_backend` can end it. Pass `null` to
+ * detach once the bridge is closed.
+ *
+ * @param bridge the bridge in use, or {@code null} to detach
+ */
+ protected void attachBridge(Bridge bridge) {
+ activeRequestRegistry.attachBridge(bridge);
+ }
+
+ /**
+ * @return whether the request currently processed by this thread has been
+ * asked to cancel via `pxf_cancel_backend`; the read/write loops poll
+ * this so they stop between fragments/records.
+ */
+ protected boolean isCancelled() {
+ return activeRequestRegistry.isCurrentCancelled();
+ }
}
diff --git
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/ReadServiceImpl.java
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/ReadServiceImpl.java
index e4558afe..d03a53e0 100644
---
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/ReadServiceImpl.java
+++
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/ReadServiceImpl.java
@@ -3,6 +3,7 @@ package org.apache.cloudberry.pxf.service.controller;
import com.google.common.io.CountingOutputStream;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
+import org.apache.cloudberry.pxf.api.error.PxfRuntimeException;
import org.apache.cloudberry.pxf.api.io.Writable;
import org.apache.cloudberry.pxf.api.model.ConfigurationFactory;
import org.apache.cloudberry.pxf.api.model.Fragment;
@@ -11,6 +12,7 @@ import org.apache.cloudberry.pxf.api.model.RequestContext;
import org.apache.cloudberry.pxf.api.utilities.Utilities;
import org.apache.cloudberry.pxf.service.FragmenterService;
import org.apache.cloudberry.pxf.service.MetricsReporter;
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestRegistry;
import org.apache.cloudberry.pxf.service.bridge.Bridge;
import org.apache.cloudberry.pxf.service.bridge.BridgeFactory;
import org.apache.cloudberry.pxf.service.security.SecurityService;
@@ -35,18 +37,20 @@ public class ReadServiceImpl extends
BaseServiceImpl<OperationStats> implements
/**
* Creates a new instance.
*
- * @param configurationFactory configuration factory
- * @param bridgeFactory bridge factory
- * @param securityService security service
- * @param fragmenterService fragmenter service
- * @param metricsReporter metrics reporter service
+ * @param configurationFactory configuration factory
+ * @param bridgeFactory bridge factory
+ * @param securityService security service
+ * @param fragmenterService fragmenter service
+ * @param metricsReporter metrics reporter service
+ * @param activeRequestRegistry registry of in-flight requests
*/
public ReadServiceImpl(ConfigurationFactory configurationFactory,
BridgeFactory bridgeFactory,
SecurityService securityService,
FragmenterService fragmenterService,
- MetricsReporter metricsReporter) {
- super("Read", configurationFactory, bridgeFactory, securityService,
metricsReporter);
+ MetricsReporter metricsReporter,
+ ActiveRequestRegistry activeRequestRegistry) {
+ super("Read", configurationFactory, bridgeFactory, securityService,
metricsReporter, activeRequestRegistry);
this.fragmenterService = fragmenterService;
}
@@ -84,6 +88,13 @@ public class ReadServiceImpl extends
BaseServiceImpl<OperationStats> implements
try {
List<Fragment> fragments =
fragmenterService.getFragmentsForSegment(context);
for (int i = 0; i < fragments.size(); i++) {
+ // stop before starting the next fragment if the request was
+ // cancelled via pxf_cancel_backend
+ if (isCancelled()) {
+ throw new PxfRuntimeException(String.format(
+ "Read of resource %s cancelled by
pxf_cancel_backend after %d of %d fragments",
+ context.getDataSource(), i, fragments.size()));
+ }
Fragment fragment = fragments.get(i);
sourceName = fragment.getSourceName();
String profile = fragment.getProfile();
@@ -148,6 +159,8 @@ public class ReadServiceImpl extends
BaseServiceImpl<OperationStats> implements
Bridge bridge = null;
try {
bridge = getBridge(context);
+ // expose the bridge so pxf_cancel_backend can end it mid-read
+ attachBridge(bridge);
if (!bridge.beginIteration()) {
log.debug("Skipping streaming fragment {} of resource {}",
context.getFragmentIndex(), context.getDataSource());
@@ -167,6 +180,9 @@ public class ReadServiceImpl extends
BaseServiceImpl<OperationStats> implements
bridge.endIteration();
} catch (Exception e) {
log.warn("Ignoring error encountered during
bridge.endIteration()", e);
+ } finally {
+ // stop exposing the now-closed bridge to
pxf_cancel_backend
+ attachBridge(null);
}
}
Duration duration = Duration.between(startTime, Instant.now());
diff --git
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/WriteServiceImpl.java
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/WriteServiceImpl.java
index f11278e9..4317522c 100644
---
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/WriteServiceImpl.java
+++
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/controller/WriteServiceImpl.java
@@ -2,10 +2,12 @@ package org.apache.cloudberry.pxf.service.controller;
import com.google.common.io.CountingInputStream;
import lombok.extern.slf4j.Slf4j;
+import org.apache.cloudberry.pxf.api.error.PxfRuntimeException;
import org.apache.cloudberry.pxf.api.model.ConfigurationFactory;
import org.apache.cloudberry.pxf.api.model.RequestContext;
import org.apache.cloudberry.pxf.api.utilities.Utilities;
import org.apache.cloudberry.pxf.service.MetricsReporter;
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestRegistry;
import org.apache.cloudberry.pxf.service.bridge.Bridge;
import org.apache.cloudberry.pxf.service.bridge.BridgeFactory;
import org.apache.cloudberry.pxf.service.security.SecurityService;
@@ -24,15 +26,18 @@ public class WriteServiceImpl extends
BaseServiceImpl<OperationStats> implements
/**
* Creates a new instance.
*
- * @param configurationFactory configuration factory
- * @param bridgeFactory bridge factory
- * @param securityService security service
+ * @param configurationFactory configuration factory
+ * @param bridgeFactory bridge factory
+ * @param securityService security service
+ * @param metricsReporter metrics reporter service
+ * @param activeRequestRegistry registry of in-flight requests
*/
public WriteServiceImpl(ConfigurationFactory configurationFactory,
BridgeFactory bridgeFactory,
SecurityService securityService,
- MetricsReporter metricsReporter) {
- super("Write", configurationFactory, bridgeFactory, securityService,
metricsReporter);
+ MetricsReporter metricsReporter,
+ ActiveRequestRegistry activeRequestRegistry) {
+ super("Write", configurationFactory, bridgeFactory, securityService,
metricsReporter, activeRequestRegistry);
}
@Override
@@ -62,11 +67,18 @@ public class WriteServiceImpl extends
BaseServiceImpl<OperationStats> implements
// dataStream (and inputStream as the result) will close automatically
at the end of the try block
CountingInputStream countingInputStream = new
CountingInputStream(inputStream);
try (DataInputStream dataStream = new
DataInputStream(countingInputStream)) {
+ // expose the bridge so pxf_cancel_backend can end it mid-write
+ attachBridge(bridge);
// open the output file, returns true or throws an error
bridge.beginIteration();
- while (bridge.setNext(dataStream)) {
+ while (!isCancelled() && bridge.setNext(dataStream)) {
operationStats.reportCompletedRecord(countingInputStream.getCount());
}
+ if (isCancelled()) {
+ throw new PxfRuntimeException(String.format(
+ "Write to resource %s cancelled by pxf_cancel_backend",
+ context.getDataSource()));
+ }
} catch (Exception e) {
operationResult.setException(e);
} finally {
@@ -76,6 +88,9 @@ public class WriteServiceImpl extends
BaseServiceImpl<OperationStats> implements
if (operationResult.getException() == null) {
operationResult.setException(e);
}
+ } finally {
+ // stop exposing the now-closed bridge to pxf_cancel_backend
+ attachBridge(null);
}
// in the case where we fail to report a record due to an
exception,
diff --git
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/rest/PxfBackendControlResource.java
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/rest/PxfBackendControlResource.java
new file mode 100644
index 00000000..c44d0c01
--- /dev/null
+++
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/rest/PxfBackendControlResource.java
@@ -0,0 +1,83 @@
+package org.apache.cloudberry.pxf.service.rest;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestRegistry;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * REST endpoints that terminate in-flight PXF requests, backing the
+ * `pxf_cancel_backend` and `pxf_interrupt_backend` SQL functions.
+ */
+@RestController
+@RequestMapping("/pxf")
+public class PxfBackendControlResource {
+
+ private static final String SEGMENT_ID_HEADER = "X-GP-SEGMENT-ID";
+ private static final String SESSION_ID_HEADER = "X-GP-SESSION-ID";
+
+ private final ActiveRequestRegistry activeRequestRegistry;
+
+ public PxfBackendControlResource(ActiveRequestRegistry
activeRequestRegistry) {
+ this.activeRequestRegistry = activeRequestRegistry;
+ }
+
+ /**
+ * Gracefully cancels the active requests of the given session on this
segment
+ * by ending their current bridge.
+ *
+ * @param sessionId the Cloudberry session id whose requests to cancel
+ * @param segmentId the originating segment id; when absent, all segments
+ * @return a JSON object `{"cancelled":N}` with the number signalled
+ */
+ @GetMapping(value = "/cancel_backend", produces =
MediaType.APPLICATION_JSON_VALUE)
+ public ResponseEntity<Map<String, Integer>> cancelBackend(
+ @RequestHeader(SESSION_ID_HEADER) int sessionId,
+ @RequestHeader(value = SEGMENT_ID_HEADER, required = false)
Integer segmentId) {
+ int filter = (segmentId == null) ? ActiveRequestRegistry.ALL_SEGMENTS
: segmentId;
+ int cancelled = activeRequestRegistry.cancel(filter, sessionId);
+ return ResponseEntity.ok(Collections.singletonMap("cancelled",
cancelled));
+ }
+
+ /**
+ * Interrupts the worker thread(s) of the active requests of the given
session
+ * on this segment.
+ *
+ * @param sessionId the Cloudberry session id whose requests to interrupt
+ * @param segmentId the originating segment id; when absent, all segments
+ * @return a JSON object `{"interrupted":N}` with the number interrupted
+ */
+ @GetMapping(value = "/interrupt_backend", produces =
MediaType.APPLICATION_JSON_VALUE)
+ public ResponseEntity<Map<String, Integer>> interruptBackend(
+ @RequestHeader(SESSION_ID_HEADER) int sessionId,
+ @RequestHeader(value = SEGMENT_ID_HEADER, required = false)
Integer segmentId) {
+ int filter = (segmentId == null) ? ActiveRequestRegistry.ALL_SEGMENTS
: segmentId;
+ int interrupted = activeRequestRegistry.interrupt(filter, sessionId);
+ return ResponseEntity.ok(Collections.singletonMap("interrupted",
interrupted));
+ }
+}
diff --git
a/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/rest/PxfStatActivityResource.java
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/rest/PxfStatActivityResource.java
new file mode 100644
index 00000000..94c1c032
--- /dev/null
+++
b/server/pxf-service/src/main/java/org/apache/cloudberry/pxf/service/rest/PxfStatActivityResource.java
@@ -0,0 +1,66 @@
+package org.apache.cloudberry.pxf.service.rest;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestInfo;
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestRegistry;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * REST endpoint that exposes the requests currently being processed by this
PXF
+ * instance. It backs the `pxf_stat_activity` SQL view.
+ */
+@RestController
+@RequestMapping("/pxf")
+public class PxfStatActivityResource {
+
+ private static final String SEGMENT_ID_HEADER = "X-GP-SEGMENT-ID";
+
+ private final ActiveRequestRegistry activeRequestRegistry;
+
+ public PxfStatActivityResource(ActiveRequestRegistry
activeRequestRegistry) {
+ this.activeRequestRegistry = activeRequestRegistry;
+ }
+
+ /**
+ * Returns the active requests known to this PXF instance, optionally
+ * filtered by the originating segment id supplied in the request header.
+ *
+ * @param segmentId the originating segment id; when absent, activity for
all
+ * segments is returned
+ * @return a JSON object `{"activities":[...]}`
+ */
+ @GetMapping(value = "/stat_activity", produces =
MediaType.APPLICATION_JSON_VALUE)
+ public ResponseEntity<Map<String, List<ActiveRequestInfo>>> statActivity(
+ @RequestHeader(value = SEGMENT_ID_HEADER, required = false)
Integer segmentId) {
+ int filter = (segmentId == null) ? ActiveRequestRegistry.ALL_SEGMENTS
: segmentId;
+ List<ActiveRequestInfo> activities =
activeRequestRegistry.snapshot(filter);
+ return ResponseEntity.ok(Collections.singletonMap("activities",
activities));
+ }
+}
diff --git
a/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/activity/ActiveRequestRegistryTest.java
b/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/activity/ActiveRequestRegistryTest.java
new file mode 100644
index 00000000..2b949692
--- /dev/null
+++
b/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/activity/ActiveRequestRegistryTest.java
@@ -0,0 +1,250 @@
+package org.apache.cloudberry.pxf.service.activity;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.cloudberry.pxf.api.io.Writable;
+import org.apache.cloudberry.pxf.api.model.RequestContext;
+import org.apache.cloudberry.pxf.service.bridge.Bridge;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.DataInputStream;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ActiveRequestRegistryTest {
+
+ private ActiveRequestRegistry registry;
+
+ @BeforeEach
+ public void setup() {
+ registry = new ActiveRequestRegistry();
+ }
+
+ private RequestContext context(int segmentId, String xid) {
+ return context(segmentId, xid, 0);
+ }
+
+ private RequestContext context(int segmentId, String xid, int sessionId) {
+ RequestContext context = new RequestContext();
+ context.setRequestType(RequestContext.RequestType.READ_BRIDGE);
+ context.setSegmentId(segmentId);
+ context.setTransactionId(xid);
+ context.setGpSessionId(sessionId);
+ return context;
+ }
+
+ @Test
+ public void emptyRegistryReturnsNoActivity() {
+
assertTrue(registry.snapshot(ActiveRequestRegistry.ALL_SEGMENTS).isEmpty());
+ assertTrue(registry.snapshot(0).isEmpty());
+ }
+
+ @Test
+ public void registeredRequestIsVisibleThenRemovedOnUnregister() {
+ registry.register(context(3, "100"));
+
+ List<ActiveRequestInfo> all =
registry.snapshot(ActiveRequestRegistry.ALL_SEGMENTS);
+ assertEquals(1, all.size());
+ assertEquals(3, all.get(0).getSegmentId());
+ assertEquals("100", all.get(0).getTransactionId());
+ assertEquals("READ_BRIDGE", all.get(0).getRequestType());
+
+ registry.unregister();
+
assertTrue(registry.snapshot(ActiveRequestRegistry.ALL_SEGMENTS).isEmpty());
+ }
+
+ @Test
+ public void snapshotFiltersBySegmentId() {
+ registry.register(context(0, "100"));
+ registry.register(context(1, "101"));
+ registry.register(context(1, "102"));
+
+ assertEquals(3,
registry.snapshot(ActiveRequestRegistry.ALL_SEGMENTS).size());
+ assertEquals(1, registry.snapshot(0).size());
+ assertEquals(2, registry.snapshot(1).size());
+ assertTrue(registry.snapshot(2).isEmpty());
+ }
+
+ @Test
+ public void concurrentRequestsTrackedIndependently() throws Exception {
+ // Each request registers/unregisters on its own thread, as the real
+ // servlet workers do; entries are tracked by ActiveRequest identity.
+ Worker w1 = new Worker(context(0, "100"), null);
+ Worker w2 = new Worker(context(0, "101"), null);
+ w1.start();
+ w2.start();
+
+ assertEquals(2, registry.snapshot(0).size());
+ w1.finishAndJoin();
+ assertEquals(1, registry.snapshot(0).size());
+ w2.finishAndJoin();
+ assertTrue(registry.snapshot(0).isEmpty());
+ }
+
+ @Test
+ public void cancelEndsBridgeAndRaisesCancelledFlagForMatchingSession()
throws Exception {
+ CountingBridge bridge = new CountingBridge();
+ Worker worker = new Worker(context(0, "100", 42), bridge);
+ worker.start();
+
+ assertEquals(1, registry.cancel(0, 42));
+ assertEquals(1, bridge.endIterationCount.get());
+
+ worker.finishAndJoin();
+ assertTrue(worker.cancelledSeen, "worker should observe the cancelled
flag via isCurrentCancelled()");
+ assertFalse(worker.interrupted, "cancel must not interrupt the worker
thread");
+ }
+
+ @Test
+ public void cancelIgnoresOtherSessionsAndSegments() throws Exception {
+ CountingBridge bridge = new CountingBridge();
+ Worker worker = new Worker(context(0, "100", 42), bridge);
+ worker.start();
+
+ assertEquals(0, registry.cancel(0, 99), "different session");
+ assertEquals(0, registry.cancel(1, 42), "different segment");
+ assertEquals(0, bridge.endIterationCount.get());
+
+ worker.finishAndJoin();
+ }
+
+ @Test
+ public void interruptWakesMatchingWorkerThread() throws Exception {
+ Worker worker = new Worker(context(0, "100", 42), null);
+ worker.start();
+
+ assertEquals(1, registry.interrupt(0, 42));
+
+ // interrupting unblocks the worker's interruptible wait on its own
+ worker.join(5000);
+ assertFalse(worker.isAlive(), "interrupted worker should have
finished");
+ assertTrue(worker.interrupted, "worker thread should have been
interrupted");
+ }
+
+ @Test
+ public void interruptIgnoresOtherSessions() throws Exception {
+ Worker worker = new Worker(context(0, "100", 42), null);
+ worker.start();
+
+ assertEquals(0, registry.interrupt(0, 99));
+
+ worker.finishAndJoin();
+ assertFalse(worker.interrupted);
+ }
+
+ @Test
+ public void cancelAndInterruptAreNoOpsAfterUnregister() {
+ // register and unregister synchronously on this thread so the entry is
+ // marked finished; both operations must then refuse to act
+ registry.register(context(0, "100", 42));
+ registry.unregister();
+
+ assertEquals(0, registry.cancel(0, 42));
+ assertEquals(0, registry.interrupt(0, 42));
+ assertFalse(Thread.currentThread().isInterrupted(), "the test thread
must not be interrupted");
+ }
+
+ /**
+ * A worker that registers itself (capturing its own thread, as the real
+ * servlet worker does), optionally attaches a bridge, then blocks on an
+ * interruptible latch until released or interrupted, and finally
unregisters.
+ */
+ private final class Worker extends Thread {
+ private final RequestContext context;
+ private final Bridge bridge;
+ private final CountDownLatch registered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ volatile boolean interrupted;
+ volatile boolean cancelledSeen;
+
+ Worker(RequestContext context, Bridge bridge) {
+ this.context = context;
+ this.bridge = bridge;
+ }
+
+ @Override
+ public void run() {
+ registry.register(context);
+ if (bridge != null) {
+ registry.attachBridge(bridge);
+ }
+ registered.countDown();
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ } finally {
+ cancelledSeen = registry.isCurrentCancelled();
+ registry.unregister();
+ }
+ }
+
+ @Override
+ public synchronized void start() {
+ super.start();
+ try {
+ assertTrue(registered.await(5, TimeUnit.SECONDS), "worker
failed to register");
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ void finishAndJoin() throws InterruptedException {
+ release.countDown();
+ join(5000);
+ assertFalse(isAlive(), "worker should have finished");
+ }
+ }
+
+ /** Minimal bridge that only counts endIteration() invocations. */
+ private static final class CountingBridge implements Bridge {
+ final AtomicInteger endIterationCount = new AtomicInteger();
+ final AtomicBoolean began = new AtomicBoolean();
+
+ @Override
+ public boolean beginIteration() {
+ began.set(true);
+ return true;
+ }
+
+ @Override
+ public Writable getNext() {
+ return null;
+ }
+
+ @Override
+ public boolean setNext(DataInputStream inputStream) {
+ return false;
+ }
+
+ @Override
+ public void endIteration() {
+ endIterationCount.incrementAndGet();
+ }
+ }
+}
diff --git
a/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/controller/ReadServiceImplTest.java
b/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/controller/ReadServiceImplTest.java
index d4ba09d0..6f3699ff 100644
---
a/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/controller/ReadServiceImplTest.java
+++
b/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/controller/ReadServiceImplTest.java
@@ -8,6 +8,7 @@ import org.apache.cloudberry.pxf.api.model.Fragment;
import org.apache.cloudberry.pxf.api.model.RequestContext;
import org.apache.cloudberry.pxf.service.FragmenterService;
import org.apache.cloudberry.pxf.service.MetricsReporter;
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestRegistry;
import org.apache.cloudberry.pxf.service.bridge.Bridge;
import org.apache.cloudberry.pxf.service.bridge.BridgeFactory;
import org.apache.cloudberry.pxf.service.security.SecurityService;
@@ -75,7 +76,7 @@ public class ReadServiceImplTest {
return result;
});
- readService = new ReadServiceImpl(mockConfigurationFactory,
mockBridgeFactory, mockSecurityService, mockFragmenterService,
mockMetricReporter);
+ readService = new ReadServiceImpl(mockConfigurationFactory,
mockBridgeFactory, mockSecurityService, mockFragmenterService,
mockMetricReporter, new ActiveRequestRegistry());
}
@Test
diff --git
a/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/controller/WriteServiceImplTest.java
b/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/controller/WriteServiceImplTest.java
index 0c5cdbc7..031eab80 100644
---
a/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/controller/WriteServiceImplTest.java
+++
b/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/controller/WriteServiceImplTest.java
@@ -4,6 +4,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.cloudberry.pxf.api.model.ConfigurationFactory;
import org.apache.cloudberry.pxf.api.model.RequestContext;
import org.apache.cloudberry.pxf.service.MetricsReporter;
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestRegistry;
import org.apache.cloudberry.pxf.service.bridge.Bridge;
import org.apache.cloudberry.pxf.service.bridge.BridgeFactory;
import org.apache.cloudberry.pxf.service.security.SecurityService;
@@ -61,7 +62,7 @@ public class WriteServiceImplTest {
});
when(mockBridgeFactory.getBridge(mockContext)).thenReturn(mockBridge);
- writeService = new WriteServiceImpl(mockConfigurationFactory,
mockBridgeFactory, mockSecurityService, mockMetricReporter);
+ writeService = new WriteServiceImpl(mockConfigurationFactory,
mockBridgeFactory, mockSecurityService, mockMetricReporter, new
ActiveRequestRegistry());
}
@Test
diff --git
a/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/rest/PxfBackendControlResourceIT.java
b/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/rest/PxfBackendControlResourceIT.java
new file mode 100644
index 00000000..5cc99db2
--- /dev/null
+++
b/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/rest/PxfBackendControlResourceIT.java
@@ -0,0 +1,92 @@
+package org.apache.cloudberry.pxf.service.rest;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.cloudberry.pxf.service.HttpHeaderDecoder;
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestRegistry;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.test.web.servlet.MockMvc;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static
org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+@WebMvcTest(PxfBackendControlResource.class)
+public class PxfBackendControlResourceIT {
+
+ @Autowired
+ private MockMvc mvc;
+
+ @MockBean
+ private ActiveRequestRegistry mockRegistry;
+
+ // required by web-layer beans pulled into the @WebMvcTest slice
+ @MockBean
+ private HttpHeaderDecoder mockHttpHeaderDecoder;
+
+ @Test
+ public void cancelBackendFiltersBySegmentAndSessionHeaders() throws
Exception {
+ when(mockRegistry.cancel(eq(3), eq(42))).thenReturn(2);
+
+ mvc.perform(get("/pxf/cancel_backend")
+ .header("X-GP-SEGMENT-ID", "3")
+ .header("X-GP-SESSION-ID", "42"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.cancelled").value(2));
+
+ verify(mockRegistry).cancel(3, 42);
+ }
+
+ @Test
+ public void cancelBackendWithoutSegmentHeaderTargetsAllSegments() throws
Exception {
+ when(mockRegistry.cancel(eq(ActiveRequestRegistry.ALL_SEGMENTS),
eq(42))).thenReturn(1);
+
+ mvc.perform(get("/pxf/cancel_backend").header("X-GP-SESSION-ID", "42"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.cancelled").value(1));
+
+ verify(mockRegistry).cancel(ActiveRequestRegistry.ALL_SEGMENTS, 42);
+ }
+
+ @Test
+ public void interruptBackendFiltersBySegmentAndSessionHeaders() throws
Exception {
+ when(mockRegistry.interrupt(eq(3), eq(42))).thenReturn(1);
+
+ mvc.perform(get("/pxf/interrupt_backend")
+ .header("X-GP-SEGMENT-ID", "3")
+ .header("X-GP-SESSION-ID", "42"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.interrupted").value(1));
+
+ verify(mockRegistry).interrupt(3, 42);
+ }
+
+ @Test
+ public void missingSessionHeaderIsBadRequest() throws Exception {
+ mvc.perform(get("/pxf/cancel_backend").header("X-GP-SEGMENT-ID", "3"))
+ .andExpect(status().isBadRequest());
+ }
+}
diff --git
a/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/rest/PxfStatActivityResourceIT.java
b/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/rest/PxfStatActivityResourceIT.java
new file mode 100644
index 00000000..20dd38cf
--- /dev/null
+++
b/server/pxf-service/src/test/java/org/apache/cloudberry/pxf/service/rest/PxfStatActivityResourceIT.java
@@ -0,0 +1,89 @@
+package org.apache.cloudberry.pxf.service.rest;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.cloudberry.pxf.api.model.RequestContext;
+import org.apache.cloudberry.pxf.service.HttpHeaderDecoder;
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestInfo;
+import org.apache.cloudberry.pxf.service.activity.ActiveRequestRegistry;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.util.Collections;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+import static
org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+@WebMvcTest(PxfStatActivityResource.class)
+public class PxfStatActivityResourceIT {
+
+ @Autowired
+ private MockMvc mvc;
+
+ @MockBean
+ private ActiveRequestRegistry mockRegistry;
+
+ @MockBean
+ private HttpHeaderDecoder mockHttpHeaderDecoder;
+
+ private static ActiveRequestInfo sampleInfo(int segmentId, String xid) {
+ RequestContext context = new RequestContext();
+ context.setRequestType(RequestContext.RequestType.READ_BRIDGE);
+ context.setSegmentId(segmentId);
+ context.setTransactionId(xid);
+ context.setGpSessionId(42);
+ context.setGpCommandCount(7);
+ context.setProfile("hdfs:text");
+ return new ActiveRequestInfo(context, 1718700000000L, "sdw3");
+ }
+
+ @Test
+ public void emptyRegistryReturnsEmptyActivities() throws Exception {
+ when(mockRegistry.snapshot(ActiveRequestRegistry.ALL_SEGMENTS))
+ .thenReturn(Collections.emptyList());
+
+ mvc.perform(get("/pxf/stat_activity"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.activities").isArray())
+ .andExpect(jsonPath("$.activities").isEmpty());
+ }
+
+ @Test
+ public void filtersBySegmentIdHeaderAndSerializesFields() throws Exception
{
+ when(mockRegistry.snapshot(eq(3)))
+ .thenReturn(Collections.singletonList(sampleInfo(3, "1234")));
+
+ mvc.perform(get("/pxf/stat_activity").header("X-GP-SEGMENT-ID", "3"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.activities[0].segmentId").value(3))
+
.andExpect(jsonPath("$.activities[0].transactionId").value("1234"))
+
.andExpect(jsonPath("$.activities[0].requestType").value("READ_BRIDGE"))
+ .andExpect(jsonPath("$.activities[0].gpSessionId").value(42))
+
.andExpect(jsonPath("$.activities[0].profile").value("hdfs:text"))
+ .andExpect(jsonPath("$.activities[0].host").value("sdw3"))
+
.andExpect(jsonPath("$.activities[0].startTimeMs").value(1718700000000L));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
