Route landing reads and updates through WTGatewayClient HASH partition pruning

[hsballoon]

【Background】
wind_tunnel_landing / landing_test use HASH(job_id) with 128 physical buckets.

WTGatewayClient.query_landing() and update_landing() can extract job_id, calculate stable_hash(job_id) % partitions, and pass the resolved bucket to dldb. Direct calls to client.session.filter() bypass this logic.

When dldb receives no explicit partitions argument for a HASH table, it iterates all existing buckets even if the SQL predicate contains job_id.

【Related Code】
core/data_manager/strategy/cloud_strategy_impl.py:
**- list_session_steps() directly calls self.client.session.filter().

• mark_latest_session_completed() directly calls self.client.session.filter().
• _load_existing_meta_json() directly calls self.client.session.filter().
• mark_records_completed() calls update_landing() using only id IN (...), without retaining or supplying the corresponding job_id.**

These paths may scan or update all landing buckets as the table grows.

【Expected Changes】

• Use WTGatewayClient.query_landing(..., as_dataframe=True) for landing reads instead of client.session.filter().
• Ensure hot landing reads and updates include job_id or pass partition=job_id.
• For mark_records_completed(), retain the associated job_id with each record ID and group updates by job/bucket.

【Notes】
No direct client.session.filter() remains in the landing hot paths above.
Queries containing one job_id pass exactly one resolved HASH bucket to dldb.
Existing result conversion, ordering and checkout_latest behavior remain unchanged.
When job_id is unavailable, the code explicitly reports/falls back rather than silently treating a full-bucket fan-out as the normal hot path.