[refactor](catalog) P5 paimon: migrate to catalog SPI + cutover (#64446)

[morningman]

Migrates the legacy in-tree Paimon catalog (fe-core datasource/paimon,
metacache/paimon, property/metastore/*Paimon* and their reverse
references) onto the catalog SPI as a self-contained fe-connector-paimon
plugin, following the maxcompute full-adopter + cutover template. paimon
is added to SPI_READY_TYPES, so a Paimon catalog now deserializes to
PluginDrivenExternalCatalog and all five functional areas — normal-table
read, system-table read, DDL, MTMV/MVCC/time-travel (procedures are a
doc-only no-op: fe-core has none) — flow through the generic PluginDriven*
bridge with no Paimon-specific branch in fe-core.

Paimon is the first lakehouse full-adopter to exercise the MVCC / sys-table /
MTMV SPI surfaces, which become the reusable template for the upcoming
iceberg/hudi migrations, so the cutover is held to byte-parity with the
legacy path. The legacy datasource/paimon/* classes are left in place (now
dead on the cutover path) and removed in a follow-up, mirroring the
maxcompute cutover -> P4 removal.

1. Generic connector-SPI surface (fe-connector-api / -spi, DefaultConnectorContext)

Source-agnostic seams so any connector can express these capabilities; every
new method is a default no-op/identity, so hive/iceberg/jdbc/maxcompute/
trino/es are byte-for-byte unaffected.

• MVCC/time-travel: resolveTimeTravel(session, handle, ConnectorTimeTravelSpec)
  • applySnapshot(...) (threads a pinned snapshot into the handle before
    planScan); new immutable ConnectorTimeTravelSpec (SNAPSHOT_ID / TIMESTAMP /
    TAG / BRANCH / INCREMENTAL) that fe-core extracts from SQL; ConnectorMvccSnapshot
    carries schemaId for schema-evolution-aware time travel.
• System tables: listSupportedSysTables / getSysTableHandle; fe-core composes
  the {base}${sys} reference name.
• Scan: proportional split-weight (getSelfSplitWeight/getTargetSplitSize),
  COUNT(*) pushdown (getPushDownRowCount, 7-arg planScan), native-read
  accounting (isNativeReadRange), getDeleteFiles (VERBOSE merge-on-read),
  ignorePartitionPruneShortCircuit (predicate-driven connectors keep genuine-null
  partitions).
• Metadata: ConnectorPartitionInfo.fileCount + SUPPORTS_PARTITION_STATS (rich
  SHOW PARTITIONS); ConnectorColumn.withTimeZone (WITH_TIMEZONE DESC marker,
  independent of the timestamp_tz mapping flag); connector cache control
  (invalidateTable/invalidateAll, schemaCacheTtlSecondOverride).
• ConnectorContext: six engine-side hooks the connector must not implement itself
  (loadHiveConfResources, vendStorageCredentials, getBackendStorageProperties,
  normalizeStorageUri x2, getStorageProperties), bound in DefaultConnectorContext
  over fe-core's single-source-of-truth credential/URI/HiveConf utilities (no
  re-ported logic that could drift).

2. Pluggable metastore SPI (NEW fe-connector-metastore-api / -spi)

The metastore-side counterpart of the fe-filesystem StorageProperties SPI: a
Paimon catalog resolves its backend (hive/dlf/rest/jdbc/filesystem) by
ServiceLoader dispatch on paimon.catalog.type instead of a fe-core switch.

• fe-connector-metastore-api: hadoop/SDK-free MetaStoreProperties contracts
  (HMS/DLF/REST/JDBC/FileSystem) carrying only Map/scalar facts.
• fe-connector-metastore-spi: MetaStoreProvider<P> + first-hit MetaStoreProviders
  dispatcher + five impls, registered via META-INF/services. HmsMetaStorePropertiesImpl
  reproduces the legacy buildHmsHiveConf key set and parity-critical ordering
  (storage overlay before the kerberos block; username/sasl last). Adding a backend =
  one provider class + one services line (no central enum/switch).
• ServiceLoader is loaded with the SPI interface's own (plugin) classloader, NOT the
  thread-context CL: at CREATE CATALOG the static init first fires on an FE worker
  thread whose TCCL is the FE app loader, so a 1-arg load would cache an empty
  provider list process-wide.
• fe-core bridges add initExecutionAuthenticator(List<StorageProperties>) so
  HDFS-Kerberos doAs is wired on the plugin path (legacy initializeCatalog is dead).

3. The Paimon connector (fe-connector-paimon)

Full read + DDL + partitions + statistics + system tables + MVCC/time-travel across
all five metastore flavors, planning native (ORC/Parquet) vs JNI reads.

• PaimonCatalogFactory (pure flavor switch: Options + Hadoop Configuration + HiveConf,
  classloader-pinned for child-first loading), PaimonCatalogOps (injection seam over
  the remote Catalog for offline-testable metadata), PaimonConnectorMetadata (~9 -> 28
  methods: schema latest/at-snapshot, sys tables, MVCC/time-travel, DDL, partitions,
  statistics, table descriptor).
• PaimonScanPlanProvider (~6 -> 30 methods): predicate/projection pushdown,
  native-vs-JNI routing, large-file sub-splitting with deletion vectors, COUNT(*)
  collapse, schema-evolution field-id dictionary, $ro unwrap to base FileStoreTable.
• New: PaimonSchemaBuilder (CREATE TABLE), PaimonIncrementalScanParams (@incr),
  PaimonTableResolver (sys/branch-aware handle -> Table), PaimonLatestSnapshotCache
  • PaimonSchemaAtMemo (per-catalog caches on the long-lived connector).
• PaimonTypeMapping gains the write direction (Doris -> Paimon) for CREATE TABLE.

4. fe-core PluginDriven host + cutover

• NEW PluginDrivenMvccExternalTable (generic MvccTable + MTMV host: the runtime class
  for paimon tables), PluginDrivenMvccSnapshot, PluginDrivenSysExternalTable +
  systable/PluginDrivenSysTable. Behavior is selected by ConnectorCapability
  (SUPPORTS_MVCC_SNAPSHOT, SUPPORTS_PARTITION_STATS) — no paimon branch in fe-core.
• PluginDrivenScanNode: MVCC pin threaded at every handle-consumption site, sys-table
  time-travel guard, COUNT pushdown, connector-agnostic EXPLAIN re-emission.
• Cutover: CatalogFactory adds paimon to SPI_READY_TYPES and drops the built-in
  case; GsonUtils removes the 7 built-in Paimon* subtype registrations and adds
  replay compat so persisted PaimonExternalCatalog (all 5 flavors) / ...Database
  -> PluginDriven* and PaimonExternalTable -> PluginDrivenMvccExternalTable on
  deserialization (existing clusters upgrade without losing catalogs).
• Nereids/DDL: PhysicalPlanTranslator drops the PaimonScanNode branch;
  UserAuthentication, CreateTableInfo, ShowPartitionsCommand generalized;
  Env.getDdlStmt renders LOCATION/PROPERTIES for SHOW CREATE (paimon engine, no
  credential leak).

5. Supporting infrastructure

• fe-filesystem: NEW HdfsFileSystemProperties / HdfsConfigFileLoader (typed HDFS;
  defaults-laden BE map vs defaults-free FE Hadoop-config map so it cannot clobber a
  co-bound object store's fs.s3a.* during a multi-backend merge); S3 MinIO support +
  legacy tuning defaults; OSS/OBS/COS ANONYMOUS provider-type; new
  FileSystemFactory.bindAllStorageProperties / FileSystemPluginManager.bindAll.
• NEW fe-kerberos leaf module: dependency-free AuthType / KerberosAuthSpec facts
  shared across modules without pulling in Hadoop.
• Packaging: NEW fe-connector-paimon-hive-shade relocates org.apache.thrift for
  HMS 2.3.7; the paimon plugin-zip excludes fe-thrift/libthrift (provided parent-first)
  to avoid a split TBase across loaders; self-contained bundle (hadoop-aws, AWS-SDK,
  hadoop-hdfs-client). paimon.version (1.3.1) pinned as the single property across
  FE connector / BE paimon-scanner / preload-extensions for FE<->BE Table/Split serde.
• BE PaimonJniScanner.getPredicates() null-predicate backstop (no-filter scan; pairs
  with the FE producer now always emitting an empty predicate list).

6. Behavior-preservation fixes vs the legacy path

Parity fixes surfaced by the SPI cutover (each restores legacy semantics):

• Storage/credentials: canonical s3.*/oss.* keys -> fs.s3a.*/fs.oss.*
  (STORAGE-CREDS); canonical AWS_* BE creds (STATIC-CREDS-BE); per-table vended
  DLF/STS token (REST-VENDED); oss/cos/obs/s3a -> s3:// for data + deletion-vector
  paths (URI-NORMALIZE); external hive.conf.resources hive-site.xml (HMS-CONFRES);
  real doAs on filesystem/jdbc + wrapped HMS read RPCs (KERBEROS-DOAS).
• Reads/types: paimon columns forced nullable (READ-NOTNULL); VARCHAR(65533) not
  widened to STRING (VARCHAR-BOUNDARY); dotted enable.mapping.varbinary /
  .timestamp_tz keys (MAPPING-FLAG-KEYS); per-type partition-value rendering
  (NATIVE-PARTVAL); genuine \N not coerced to NULL (PARTITION-NULL-SENTINEL);
  field-id history dict for rename-safe native reads (SCHEMA-EVOLUTION); paimon-native
  split encode under enable_paimon_cpp_reader (CPP-READER).
• Planning/perf: restore the force_jni_scanner escape hatch (FORCE-JNI-SCANNER);
  intra-file native sub-splitting (NATIVE-SUBSPLIT); precomputed COUNT(*) row count
  (COUNT-PUSHDOWN); table row count for CBO / SHOW TABLE STATUS (TABLE-STATS).
• Time travel / DDL: FOR TIME AS OF under CST/PST session zones (TZ-ALIAS); resolved
  • validated driver_url for jdbc catalogs (JDBC-DRIVER-URL); reject a local-only
    name conflict on CREATE TABLE (CREATE-TABLE-LOCAL-CONFLICT).

7. Doc-sync

plan-doc P5 design / task-list / HANDOFF / deviation tracking notes.

Tests & verification

• ~70 new FE unit suites: paimon connector (offline recording fakes:
  RecordingPaimonCatalogOps / FakePaimonTable), fe-core PluginDriven* MVCC /
  sys-table / scan-node / split-weight, metastore-spi dispatch + per-backend
  properties, fe-filesystem HDFS/MinIO/anonymous, fe-kerberos.
• checkstyle 0 violations; connector import-gate passes; FE main+test compile clean.
• End-to-end paimon behavior is docker-gated (enablePaimonTest=true); the five
  functional areas are covered by the external_table_p0/paimon regression suites,
  green on the latest CI run for this branch.

[hello-stephen]

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?

[924060929]

Part of a systematic post-merge review of the catalog-SPI series (#65185), verified against branch tip 3ba75b7 and legacy code on current master. Positive note first: GSON/persistence compat, sys tables, SHOW PARTITIONS columns and property passthrough all checked out clean in a separate removal audit, and three scary-looking predicate-converter candidates were REFUTED during adversarial verification because supportsCastPredicatePushdown()=false strips any cast-bearing conjunct before conversion — that guard is load-bearing (the value.toString() / Number-truncation arms in PaimonPredicateConverter are only safe because of it), so consider pinning it with a test/comment so a future 'enable cast pushdown' change doesn't silently unleash those arms.

Findings that anchor outside this PR's hunks:

1. convertAnd partial-conjunction divergence (PaimonPredicateConverter.java:121): it keeps a partial conjunction of the convertible children, while legacy required BOTH sides of a nested AND to convert (master PaimonPredicateConverter.java:75-80 returns null otherwise). Verified sound — a partial AND is strictly weaker, convertOr rejects any unconvertible disjunct, and BE re-applies the full predicate — so it only changes which files are pruned (plan/profile-visible, strictly better). Worth a comment marking the divergence deliberate so a future parity audit doesn't re-flag it.

2. selectedPartitionNum semantics changed (PluginDrivenScanNode): it now reports the Nereids FE-pruned partition count, while legacy paimon reported distinct partitions among SDK-planned splits (master PaimonScanNode.java:502). Both EXPLAIN partition=N/M and the sql_block_rule partition_num check (StmtExecutor passes scanNode.getSelectedPartitionNum() to SqlBlockRuleMgr for FileScanNodes) observe this — a rule that passed on legacy (limit 5; SDK touches 3 of 10 FE-selected partitions) now blocks the query.

3. Shared infra placed in the connector: buildHadoopConfiguration/assembleHiveConf (storage-config assembly + classloader pinning) live as statics inside PaimonCatalogFactory; the iceberg migration already copy-pasted them (IcebergCatalogFactory javadoc literally says 'mirror PaimonCatalogFactory'). Every classloader/auth fix now needs manual re-application per connector — fe-connector-metastore-spi looks like the natural shared home.

[924060929]

Adding paimon here is the line that flips paimon EXPLAIN output from VPAIMON_SCAN_NODE to VPluginDrivenScanNode (verified live on this branch; master PaimonScanNode.java:159 passes "PAIMON_SCAN_NODE"). The tracking issue's invariant #3 explicitly lists EXPLAIN output as preserved, and plan-doc/deviations-log has no entry for the header rename (the fix-e work restored the explain body lines byte-for-byte but accepted the renamed header). If the unified name is intentional, suggest recording it as a deviation on #65185; if not, planNodeName = connectorType.toUpperCase() + "_SCAN_NODE" in PluginDrivenScanNode's ctor preserves the legacy header for every migrated connector at one stroke.

[924060929]

file_format here comes from the table-level file.format option (default parquet), while legacy derived it from the first data file's suffix with the option only as fallback (master PaimonScanNode.java:259 + PaimonSplit.java:59) — and the native path in this file still does suffix derivation. For a table whose files are ORC but whose options omit file.format (or whose format changed after writes), JNI/COUNT ranges now claim parquet. The only consumer appears to be paimon_cpp_reader's FILE_FORMAT/MANIFEST_FORMAT backfill, so this needs either an e2e check with an option-unset ORC table on the paimon-cpp path, or simply restoring suffix derivation for these two builders to match the native path.

[924060929]

The ignore_split_type session variable (IGNORE_JNI / IGNORE_NATIVE debug filtering) is not consumed anywhere on the plugin path — legacy honored it at four points in PaimonScanNode.getSplits (master :368-369, :389, :431, :471), and the variable still exists in SessionVariable at tip, so SET ignore_split_type=ignore_jni now silently no-ops for paimon. Either wire it into this routing loop or deprecate the variable; a silently dead debug knob is the worst of both.

[924060929]

The to-Paimon (CREATE TABLE) direction drops nested nullability and struct-field comments: ARRAY element / MAP value / ROW fields get default nullability and no comment, while legacy DorisToPaimonTypeVisitor copied containsNull per element/value/field and carried StructField comments (master :57-78). Notably this is not an SPI limitation — ConnectorColumnConverter carries childrenNullable + childrenComments and IcebergSchemaBuilder consumes them; this connector just never reads them. Tables created through Doris now get a different paimon schema than legacy produced (visible to Flink/Spark and SHOW CREATE).

[924060929]

spec.getInitialValues() is never read, so CREATE TABLE ... PARTITION BY LIST (col) (PARTITION p VALUES IN ('x')) silently accepts and drops the explicit partition value definitions (verified live on this branch — table created, definitions gone). Legacy also dropped them, so it's not a regression, but the SPI now models the values (ConnectorPartitionValueDef / getInitialValues, javadoc: 'only meaningful for LIST/RANGE') while the only producer hardcodes Collections.emptyList() — dead API surface plus DDL accepted that cannot be honored. Since this builder already fail-fasts on unsupported transforms a few lines below, rejecting non-empty initial values with a clear error would be consistent and honest.

[924060929]

This is now the third fe-core string switch that must be hand-edited per connector (with PluginDrivenExternalTable.getEngine() and getEngineTableTypeName()); the javadoc itself says they 'must stay in sync'. P6 already added its iceberg cases to all three, confirming the N-connectors × N-edits trajectory. Connector identity (legacy engine name / table-type name) looks like it belongs on the Connector SPI (e.g. Connector.getLegacyEngineName()), declared once per plugin — otherwise every future connector risks the exact silently-wrong-engine-name regression these switches exist to prevent, with no compile-time guard tying them together.