wwj6591812 opened a new pull request, #9111: URL: https://github.com/apache/paimon/pull/9111
## Background and motivation Paimon's existing query service is designed for primary-key, fixed-bucket tables. Data-evolution append tables have no primary key and normally use `bucket = -1`; pretending that a lookup column is a primary key, or routing every request to physical bucket 0, is therefore incorrect. This change adds an explicit remote exact-key query service for append tables whose application contract makes one non-null column unique. The initial use case is a large append-only image Blob table keyed by a normalized object name. Returning only the Blob descriptor lets an inference service reuse the original append table without maintaining a second descriptor primary-key table. ## Approach A BTree global index is used as an exact coverage gate for one pinned snapshot. Requests do not point-read the BTree. 1. One monitor selects and leases an exact snapshot. 2. It validates complete BTree coverage and plans that snapshot's splits. 3. Bootstrap subtasks read each split once and project only lookup/value fields. 4. Rows are shuffled with `floorMod(BinaryRow.hashCode(), parallelism)`. 5. Each executor builds shadow local state and starts a bounded query server. 6. Discovery becomes READY only after every shard acknowledges the same generation and snapshot. ## Correctness and consistency - The feature does not create or emulate a primary key. - It accepts only unpartitioned, main-branch, bucket-unaware data-evolution append tables. - Keys are canonicalized to `RowKind.INSERT`. - Null table keys are skipped; null request keys are rejected. - A duplicate non-null key invalidates the whole generation. There is no first/last-wins policy. - An unserveable or oversized individual value invalidates the generation before READY. - Seeing a new generation immediately raises every executor's accepted fence. A cached old descriptor therefore gets STALE/NOT_READY instead of a false MISS for an unindexed tail. - The client validates server epoch, generation, snapshot ID and snapshot identity before interpreting values or structured errors. - `getValuesWithMetadata` exposes the served fence. A MISS is exact for the advertised snapshot; it is not an implicit read-after-write guarantee. ## Blob and compaction semantics A selected top-level raw Blob field is read with `blob-as-descriptor=true` inside the query-service table copy. State and responses contain `BlobDescriptor.serialize()`, never Blob payload bytes. Nested Blob, inline descriptor and BlobView fields are rejected in this version. Ordinary and Blob compaction create a new fenced generation. A per-attempt snapshot lease protects active/building snapshots and provides an acknowledged handover grace period so in-flight descriptors remain readable. Disabling Blob compaction is a reasonable first-deployment optimization, but correctness does not depend on it. ## Discovery and protocol Protocol v2 discovery contains table UUID, branch, schema fingerprint, ordered lookup/value field IDs, generation, snapshot identity, layout/hash version, owner token, readiness and one address/server epoch per shard. Expected failures use typed, bounded normal response frames. This client never deserializes legacy Java `Throwable` frames. Requests and responses enforce key-count, byte and frame limits; requests have a timeout and one discovery-refresh retry; servers use a bounded queue and return typed OVERLOADED responses. ## Recovery State is ephemeral and rebuilt after restart. New attempts first publish NOT_READY with new server epochs. Owner-scoped descriptors and tombstones prevent delayed attempts from overwriting, deleting or reviving newer owners. Consumer leases survive task close until persisted expiration, protecting failover handover. READY and exact all-shard NOT_READY acknowledgements each get independent grace deadlines, so continuous commits cannot permanently pin the oldest snapshot. ## Compatibility The existing primary-key QueryService, KV protocol and two-argument Procedure remain unchanged. The append path is selected only when lookup/value arguments are present. Flink 1.18 gets a compatible six-position overload. The new remote wrapper is explicit and does not change SQL lookup-join selection. ## Security and operational boundary The service rejects query-auth-enabled tables and unsafe corrupt/lost-file or consumer options, checking persisted schema options as well as dynamic copies. Consumer IDs are bounded safe path segments. This transport does not provide end-user authentication or TLS. Executor ports must stay on a trusted network, with an authenticated Gateway for external traffic. This version performs one full projected scan per fully indexed snapshot and stores one projected value per key on executor-local disk. Parallelism distributes state and requests, but refresh still incurs a full scan and shuffle. Operational constraints: - Run only one logical service job for one lookup projection; owner sequence allocation is not a cross-job CAS. - Service FileIO needs read-after-write visibility and consistent listings. - The lookup key must be unique for all non-null values. - New snapshots need complete BTree coverage before READY. - Persisted `consumer.expiration-time` must exceed lease grace plus recovery margin. ## Tests - Core query/coverage/location/lease/ServiceManager: 35 tests. - Service protocol/network/runtime: 34 tests. - Flink monitor/register/action/procedure: 12 tests. - MiniCluster global-index E2E: 3 tests covering hit/miss/batch, unindexed tail, refresh, stale fences, Blob descriptor/compaction, public Action/Procedure, remote wrapper, same-job TaskManager failover, cancellation tombstone and P1-to-P2 rescale. - Existing PK RemoteLookupJoin: 2 tests. - Flink 1.18 legacy and six-argument positional Procedure tests passed. - Flink 2.2/JDK11 reactor package and five-module validation passed. Flink 1.17/VVR8 is intentionally a downstream backport/release gate. ## Suggested review order The branch is split into three commits: 1. Core query, coverage, lease and discovery contracts. 2. Transport hardening and protocol framing. 3. Flink topology, public entry points, integration tests and documentation. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
