IMPALA-14523: JDBC Scan Parallelism with Shared Cursor Fetching

Impala’s current JDBC DataSourceScan creates a separate JDBC
connection per scanner thread, but since JDBC cursors are
inherently sequential, this limits effective parallelism and
introduces unnecessary connection and JNI overhead. As a result,
even with MT_DOP enabled, scans do not scale efficiently and can
suffer from poor throughput for large datasets.

This patch introduces a shared JDBC connection model per fragment,
allowing multiple scanner threads to cooperatively fetch from a
single cursor. A new SharedJdbcConnection abstraction is added and
owned by DataSourceScanPlanNode, ensuring the connection is opened
and closed exactly once. Multiple DataSourceScanNode instances
invoke FetchBatch() concurrently, where cursor movement is
serialized while row materialization happens in parallel, and EOS
is tracked via an atomic flag to avoid redundant fetches.

Planner and scheduler changes ensure all JDBC scan ranges are
pinned to a single executor using a new is_data_source_scan flag,
and one virtual scan range is generated per MT_DOP instance. On the
Java side, batch fetching with minimal locking is introduced, and
JdbcDataSource is updated for safe concurrent access and improved
materialization, reducing overhead and improving scan throughput.

For instance, for query:
WITH recent_orders AS (
  SELECT o.o_orderkey, o.o_custkey, o.o_totalprice, o.o_orderdate
  FROM tpch_jdbc.orders o WHERE o.o_orderdate >= '1995-01-01'
)
SELECT c.c_custkey, c.c_name, c.c_mktsegment,
       COUNT(ro.o_orderkey) AS num_orders,
       SUM(ro.o_totalprice) AS total_spent,
       AVG(ro.o_totalprice) AS avg_order_value,
       RANK() OVER (ORDER BY SUM(ro.o_totalprice) DESC) AS spend_rank,
       CASE
         WHEN SUM(ro.o_totalprice) > 500000 THEN 'VIP'
         WHEN SUM(ro.o_totalprice) > 100000 THEN 'Premium'
         ELSE 'Regular'
       END AS customer_tier,
       s.s_name AS supplier_name,
       s.s_acctbal AS supplier_balance
FROM tpch_jdbc.customer c
JOIN recent_orders ro
     ON c.c_custkey = ro.o_custkey
JOIN tpch_jdbc.supplier s
     ON c.c_nationkey = s.s_nationkey
GROUP BY c.c_custkey, c.c_name, c.c_mktsegment, s.s_name, s.s_acctbal
ORDER BY total_spent DESC
LIMIT 5;

It takes around 260 sec with mt_dop =1 while with mt_dop =7,
it takes close to 77 sec.

Also added JdbcJniWaitTime, JdbcCursorFetchTime,
JdbcLockWaitTime and JdbcJniCallCount for better analysis
of jdbc queries, in query profile.

Testing:
Some planner tests are included in jdbc-parallel.test and benchmark
tests are added in test_ext_data_sources.py.

Generated by:
Some codes and comments are generated with
the help of claude-4.6-sonnet-medium-thinking.

Change-Id: I3b25b99f5cb77d32c111ba37c0a01378ffdc1107
Reviewed-on: http://gerrit.cloudera.org:8080/24133
Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
Reviewed-by: Zoltan Borok-Nagy <boroknagyz@cloudera.com>