[feat](fluss) Support reading Apache Fluss tables through a fluss catalog - #66399
Open
morningman wants to merge 35 commits into
Open
[feat](fluss) Support reading Apache Fluss tables through a fluss catalog#66399morningman wants to merge 35 commits into
morningman wants to merge 35 commits into
Conversation
Introduces fe-connector-fluss: the plugin identity (provider, services entry, plugin zip), the catalog property contract, and FlussAdminOps -- the single interface through which this connector talks to a fluss cluster. FlussAdminOps exists so that metadata mapping and, later, split generation stay pure functions of what the cluster reports. That is what lets a unit test drive the states that decide correctness (an empty bucket, a lake offset that caught up with the log, a partition that only the lake still has) without a mocking framework, which the fe-connector modules do not carry. Notes on the fluss client that shape the pom: - it ships as one unrelocated fat jar carrying fluss-common/-rpc, frocksdbjni, commons-lang3 and zstd, so fluss-common must NOT be declared separately and frocksdbjni cannot be excluded; - it declares Arrow `provided`, so a consumer reaching fluss's ARROW log format has to add Arrow itself. Metadata and planning do not, so this module does not. Fluss errors travel unwrapped: scan planning has to discriminate on LakeTableSnapshotNotExistException to decide whether a union read is possible, and wrapping would turn that into string matching. The TCCL pin sits at connection creation because that is where the client's own threads are born and inherit a context classloader; pinning once there covers them for life. Version is pinned by one property in fe/pom.xml for every fluss consumer: the FE plans splits against the same client the BE scanner will read with, and the scanner links fluss classes marked @internal, so a skew breaks the read path at runtime rather than at build time. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Adds FlussTypeMapping, a DataTypeVisitor over fluss's 19 type roots, so a type added by a future fluss release breaks this compile instead of degrading a column silently. Every rule is the composition of fluss's own FlussDataTypeToPaimonDataType with the paimon connector's PaimonTypeMapping: a datalake table is readable both as `tbl` (this mapping) and as `tbl$lake` (delegated to the paimon connector), and one table must not show two schemas. That is what fixes CHAR over 255 collapsing to STRING, the microsecond clamp on timestamps, and the spelling and defaults of the two mapping switches. TIME is marked UNSUPPORTED rather than reinterpreted: Doris has no storable TIME column, and both substitutes other engines use (STRING, elapsed-millis INT) hand back a value whose meaning differs from the source. The marker degrades one leaf, so a wide table with one TIME field still loads. MAP is mapped, not marked unsupported. Fluss reads it in the ARROW log format, in the compacted KV format behind primary-key tables, and converts it on the way into the lake, so refusing it here would only mean `tbl$lake` showing a MAP that `tbl` refuses to project. The two mapping switches reuse the unprefixed, engine-wide names the hive, paimon and iceberg catalogs already answer to; being unprefixed also keeps them out of the fluss client configuration for free. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Implements the metadata surface: the table handle, the Doris schema and column handles, partition listing, statistics and the table descriptor. The handle snapshots what split planning will need — primary key, bucket count and keys, partition keys, lake format, and the fluss table property map, which for a datalake table is where the fluss coordinator merges the lake catalog's own connection settings (the Doris catalog is configured with bootstrap servers and nothing else). Taking them one at a time later could straddle an ALTER; taking them together cannot. Column comments are read from the fluss schema, not from the table's row type: fluss rebuilds that row type from the schema and drops every field description on the way, so reading the row type would report every column as undocumented. Columns are reported nullable throughout even though fluss marks primary-key columns NOT NULL — propagating it would let the planner fold null-rejecting predicates on this path while the same table read through its lake sibling (the paimon connector, which reports everything nullable) keeps them. Partitions are rendered in the Hive-style `k=v/k=v` naming fe-core parses back, not fluss's own `v$v`. No escaping is needed: fluss rejects any partition value outside ASCII alphanumerics, `_` and `-`, so no value can contain the separators, and none can be SQL NULL. One statement reads a table once. The handle, the schema, the column handles and the comment share a per-statement memo, which beyond the round trips is what stops a concurrent ALTER from being visible to half of one plan. Tests come in two layers. The unit layer drives the awkward states a cluster will not produce on request — a table that vanished, a coordinator that is down, a partition spec whose map iterates in a different order than the partition columns. Over it, FlussMetadataClusterTest starts a real coordinator, tablet server and ZooKeeper in the test JVM and reads tables created through the fluss client, which is the only way to check the premises the unit layer assumes. It is named ...Test rather than ...ITCase because surefire's default includes do not match *ITCase, and that name would leave the class unexecuted under a green build. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The regression suites need a fluss cluster to read from. This brings up ZooKeeper, a fluss coordinator and tablet server, and a Flink cluster whose SQL client creates the fixture tables once and then idles, so that `compose up --wait` gates on the fixtures being complete rather than on the containers merely running. Fluss 1.0 is not released, so neither image can be pulled: build-images.sh builds both from a local fluss checkout, which is also why the component is not part of the default component set yet. Two things the servers need because Doris runs outside the compose network: the advertised listeners are the host address plus the published port, and remote.data.dir is bind mounted at the same absolute path on both sides, since BE will read the kv snapshots written there directly. Also adds fluss to both connector lists in build.sh. They were missing it, so the plugin was never packaged or deployed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A fluss scan range is one bucket of one partition, read as a log range, a full primary-key range, or a primary-key range unioned with the lake. This adds the range itself and the two thrift maps that carry it. The payload is an untyped string map rather than a thrift struct of its own. The C++ layer holds no fluss logic -- it hands the map straight to the java scanner in be-java-extensions -- so a struct would only add a field-by-field transcription in the middle, which is what paimon pays for in paimon_jni_reader.cpp. es_params and jdbc_params are the same shape for the same reason, and it keeps a plugin's parameters out of the core IDL. Only what varies per split rides in the range. Bootstrap servers, table identity and the client and table options are the same for every split of a scan, so they go once into TFileScanRangeParams.fluss_properties, next to the paimon and ES properties already there for that reason; a table with 100 partitions and 128 buckets would otherwise serialize them 12800 times. Nothing populates the scan-level map yet -- the split planner does, in the next change. Partition columns are declared to the engine rather than returned by the scanner. That is forced by the union read: the lake half of a union is paimon ranges, the paimon connector already declares its partition keys that way, and the file-slot / partition-slot split is decided once per scan node. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Reads a log table one bucket at a time, from the earliest offset fluss still holds up to the offset the log had reached when planning ran. That stopping offset is taken once per partition so every bucket of a partition stops at the same view of the table; a bucket nobody has written to yields no range at all. Partition naming moves into FlussPartitions, shared with the metadata listing. The pruned partition names the engine hands to planning are the ones that listing produced, so the two have to render them identically -- rendered separately, planning matches none of them and scans nothing, which looks like an empty table rather than a bug. What this cannot serve yet it refuses by name: a primary-key table, and a table whose tiering has committed to the lake. Reading the latter as fluss-only would return whatever the log still holds and drop everything already tiered away -- a query that succeeds with rows missing. fluss.union_read.mode=disabled is how a user asks for that read on purpose, and a lake table that has never tiered still falls back to the log alone, because then the log is the whole table. Planning only reads metadata, so it is safe under EXPLAIN, which does reach planScan. The EXPLAIN line it appends is what a regression test has to read to tell a union read from a silent fallback to a fluss-only one. The cluster tests that write rows need --add-opens java.nio for Arrow, fluss's default log format; the BE JVM that will host the scanner already gets that flag from bin/start_be.sh. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Turns one scan range -- one bucket of one partition over a bounded offset range -- into rows in Doris's vector table, by driving a fluss log scanner and converting each fluss row column by column. Where the read stops needs all three of fluss's own conditions, taken from its bounded reader in KvSnapshotAndLogBatchScanner: drop a record at or past the stopping offset, stop right after the record before it, and stop when the fetch has consumed up to the stopping offset without yielding a record there. The second exists because the record AT the stopping offset may never be written -- that offset is where the log had got to, not a row -- so polling on would block forever. The third covers a tail of control records, which take offsets but are never handed to a scanner. Byte-valued columns dispatch on the FLUSS type, not the Doris one. The two do not line up: fluss BYTES maps to Doris STRING unless the catalog opts into VARBINARY, so one column may be asked for either way, and CHAR and BINARY are stored fixed-width and read back as nothing without their declared length. fluss-client already carries a relocated Arrow, allocator included, so nothing here declares one; the arrow coordinates its pom marks provided are not the ones it uses. What it does need is the java.nio add-opens, which bin/start_be.sh already passes. Tests run against a real cluster because both ends of the conversion are someone else's format. They read values back through the STRING mapping rather than VARBINARY: java-common's getBytesVarbinary decodes a StringView layout that neither appendVarbinary nor BE's _fill_varbinary_column writes, so that read-back path cannot check a byte value. Production is unaffected -- the write side and BE agree. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
FileScannerV2 gains a fluss table format. The reader holds no fluss logic: it merges the scan-level fluss_properties with the per-range fluss_params - the range wins, being the specific half - and hands the result to org.apache.doris.fluss.FlussJniScanner. Partition columns are left out of what the scanner is asked for. FE declares them as path_partition_keys and ships their values on each range, so reading them per row would repeat what the split already states once, and would lay the block out differently from the paimon half of a union read. Only FileScannerV2 is wired. With enable_file_scanner_v2=false the legacy scanner reports "Not supported create reader for table format: fluss", so a fluss regression suite has to pin the session variable - the fuzzy session mode randomizes it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
Reads the docker fixtures through a fluss catalog: whole table, projection in a different order than the schema, predicates, every mapped type including the all-NULL row, and the partition columns BE materializes from each range rather than reads. Two shapes the planner has to get right are pinned through the plan: pruning a partition must shrink the ranges, not just the partition=N/M line, and a table that was never written to must plan no range at all. The fixture set gains an empty table for the latter. Values are asserted explicitly instead of through a .out baseline, so an expectation and the fixture that produces it stay in one file. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
The build cache extension takes a dependency's type for its file extension when it hashes a module's inputs, so type "test-jar" sent it looking for a fluss-server-1.0-SNAPSHOT.test-jar that exists in no repository and the frontend build failed before compiling anything. The classifier spelling resolves the very same file. These were the only two modules in the tree depending on a foreign test-jar, which is why nothing caught it earlier: the module tests were always run with the build cache turned off. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
A query over a partitioned table can need none of the columns the scanner reads: "select dt, count(*) ... group by dt" wants only the partition column, and BE materializes that one from the range itself. Fluss rejects an empty projection outright, so opening the scanner threw and the query failed. Stand the narrowest legal request in for it; nothing reads the column that comes back, since getNext iterates over the required fields and reports the row count alone. count(*) alone did not reach this: the planner keeps the first column as a placeholder slot, so only a projection made entirely of partition columns empties out. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
A fluss primary-key table cannot be read the way a log table is: its log is a change log, so replaying it verbatim returns superseded and deleted rows. Each bucket is planned instead as its latest kv snapshot plus the change log that followed it, which the scanner merges by key. A bucket fluss has never snapshotted carries -1 and the earliest sentinel, and its state is rebuilt by replaying the whole change log -- equally correct, only slower. Snapshots are asked for before offsets, and the order is load-bearing: a snapshot committed between the two calls ends past the offset planning stopped at, and that bucket would then be read from a snapshot already containing rows written after the query started while every other bucket stopped where planning saw it. Log offsets only move forward, so this order keeps every snapshot at or behind the stopping offset. EXPLAIN counts primary-key ranges apart from log ranges. The two are read by different code, and a single total cannot tell a primary-key table planned the right way from one planned the wrong way. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
A primary-key table's log is a change log, so replaying it verbatim returns superseded and deleted rows. A PK_FULL range is read instead through fluss's own KvSnapshotAndLogBatchScanner, which merges the bucket's kv snapshot with the change log that followed it, by key. Both reads are now fluss BatchScanners -- the bounded log read moves into one, carrying its three stop conditions verbatim -- so the row loop no longer knows which kind of range it is draining. The projection differs between them and that is deliberate: the log path still needs a stand-in column when the projection is empty, because fluss rejects an empty one, while the primary-key reader appends the key columns to what it fetches and projects the merged row back down, so it never asks for an empty projection at all. The C++ glue needs no change: it merges the two parameter maps and forwards them, so fluss.kv_snapshot_id reaches the scanner without anything in between knowing what it means. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
…back A fluss primary-key table with no kv snapshot is read by replaying its whole change log, which is correct and takes a completely different code path from the one this environment exists to cover: BE reads snapshot FILES that the fluss container wrote, from the host, at an absolute path both sides share. With the default ten-minute interval whether a suite exercised that path was down to when it happened to run. The server now snapshots every ten seconds and startup does not report the environment ready until every primary-key fixture has one on disk. That does not pile up files -- a tablet whose log has not advanced since its last snapshot is skipped -- and the wait proves a snapshot was taken, not that the coordinator has committed it (completion is registered in ZooKeeper and the directory here is created before the upload). Losing that millisecond costs a suite nothing: the read falls back to the change log and still returns the right rows. Adds pk_part, a partitioned primary-key table with an update in one partition and a delete in another: a partitioned table is snapshotted per partition, so a merge that crossed partitions loses or resurrects one of them. KNOWN FAILING, and the reason is not in this commit: reading a kv snapshot inside BE aborts the process. The RocksDB JNI library that fluss-client bundles and doris_be define 2576 rocksdb symbols under identical mangled names, and the executable wins the lookup, so the library's calls land in a different RocksDB built against a different libstdc++ string ABI. See plan-doc/HANDOFF.md. The same scanner code passes 10/10 against a real cluster in a plain JVM, which is the A/B. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
Exporting them makes this executable the definition every later-loaded library binds to, so a JNI library carrying its own RocksDB runs half on ours. The fluss scanner bundles frocksdbjni, whose librocksdbjni.so defines 2576 rocksdb symbols under names identical to ours but was built against the pre-C++11 libstdc++ string ABI: objects laid out by one copy and used by the other yield a garbage length, an std::bad_alloc that escapes the JNI frame, and an aborted BE. Reading any fluss primary-key table with a kv snapshot killed the process, reproducibly. Scoped to the archive rather than dropping ENABLE_EXPORTS, because what needs the exports is native UDFs (runtime/user_function_cache.cpp dlopens them) and those use the Doris UDF ABI, which has nothing to do with RocksDB. Crash stacks do not need it either -- they are symbolized from debug info, which is why they name even anonymous-namespace functions. 61 rocksdb symbols remain exported: inline and template members the compiler emitted into Doris's own objects, which no archive exclusion can reach. 29 of those still share a name with the JNI library, but none appear in its relocation table -- it never resolves them at load time, so they cannot be interposed. The library also duplicates zstd, lz4, snappy, bzip2 and zlib symbols; those are C ABIs, stable and layout-free, and are left alone. Verified: the fluss primary-key suite passes with BE alive (it aborted before); all three fluss suites green; an internal table survives write, BE restart and read, which is the tablet metadata RocksDB itself round-tripping. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
A datalake-enabled fluss table now exposes a "lake" system table, so tbl$lake reads the paimon table its tiering service writes. The read is delegated whole: the handle comes from an embedded paimon sibling connector, the engine routes that handle's scan to the sibling's own planner, and this connector's metadata forwards every per-handle call. The sibling is built through ConnectorContext.createSiblingConnector, so the fluss plugin bundles no paimon class at all -- the same contract the hive gateway uses for its iceberg and hudi tables, and the reason its zip is unchanged at four artifacts. Its catalog properties come from the fluss TABLE's properties, where the coordinator injects the cluster's datalake.paimon.* settings; only a filesystem catalog is served for now, and anything else fails loud rather than half-configuring a catalog. Exactly one sibling per catalog, which is a correctness constraint and not thrift: two paimon siblings answer "is this handle yours?" with the same class test, so a second one could never be routed apart from the first and a table would silently read the wrong warehouse. A second lake configuration is refused with a message asking for a catalog refresh. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
The engine pins the context classloader to the plugin whose object it is about to call, so everything the fluss connector runs sees the fluss plugin's loader. That is right until the connector calls its lake sibling: the sibling's SDK is loaded child-first by ITS plugin, and the parts of it that discover implementations by ServiceLoader — catalog factories, file IO, file formats — look them up through the context classloader, where none of them exist. The failure would surface as a NoClassDefFoundError at the first lake table read rather than at wiring time. Route every forward through one helper that pins the loader and restores the caller's, including on a throw. Building the sibling's metadata is inside the pin too: that call already opens the lake catalog. The helper also owns the per-statement metadata memo, so the pin and the shared instance cannot be forgotten by a new caller. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
A table that fluss tiers into a lake was refused outright: reading it as fluss-only would return just the rows tiering has not moved yet, which is a successful query with rows missing. It is now read as the union of the two halves — the lake at the snapshot fluss reports as readable, plus each bucket's log from exactly the offset that snapshot recorded, so the halves meet with no row read twice and none skipped. The lake half is not planned here. It is planned by the paimon sibling connector this catalog already builds for `tbl$lake`, on a handle pinned through the SPI's own snapshot hook, and its ranges are mixed into the same scan node — BE builds a reader per range, so the two kinds coexist. That keeps the plugin free of any paimon dependency (its zip is unchanged: connector jar, fluss-client, fe-foundation, jsr305) and gives the lake half paimon's native readers, deletion vectors and file cache for free. Nothing paimon-specific is named: the pin is a snapshot id with no connector options, and what fluss records as the lake snapshot IS the id the lake returned when tiering committed it. Three things the halves must agree on. The lake snapshot is read BEFORE the log's stopping offsets, so it can only be at or behind them; asked the other way round, a snapshot committed in between would cover rows past where the log half stops. The lake half is planned on the LAKE's own column handles — the sibling projects by its own handle type and ignores anything else, so fluss's handles would leave it reading every column, the three system columns tiering appends included. And the node properties of both halves travel in one map, since the engine populates the scan params once; the shared keys must already be equal, so a difference is raised rather than resolved by picking a side. A primary-key table tiered into a lake is still refused: its two halves have to be merged by key rather than concatenated. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
Connector.ownsHandle defaults to false, and this connector never overrode it. That was invisible while paimon was only ever a front-door catalog: the predicate exists so a GATEWAY connector can embed another as a sibling and route a foreign handle back to whoever made it, since the sibling's concrete handle type cannot be named across the plugin classloader split. The fluss connector reads a lake table by delegating to this one, so it asks that question about every handle it gets back — and got "not mine" about handles paimon had just produced. Every guard on the gateway side then falls through, and the first cast throws a ClassCastException naming the GATEWAY's handle type and two class loaders, with nothing to suggest the missing piece is a method here. Same one-liner the iceberg and hudi siblings behind the hms gateway already carry. No unit test could have caught this: a hand-written test double implements ownsHandle precisely because it has to, so the double is more capable than the real connector. It took an end-to-end read to surface. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
FileScannerV2 built its table reader once, from the first range, and reused it for every range after that. One scan node can be given ranges of more than one table format: a fluss union read plans the table's lake half through the paimon connector and its log half itself, and both arrive as ranges of the same scan. Whichever range came first then decided the reader for all of them, and the other format's ranges were handed to a reader that does not understand them. That does not fail cleanly — it fails as whatever that reader makes of a foreign range. Here it was paimon's, reporting an unsupported file format for a fluss range that carries no paimon parameters at all. Which ranges share a scanner is up to the engine's assignment, so the same query succeeded or failed by how the ranges happened to be dealt out, and changing the projected columns could flip it either way. The reader now follows the range's table format. The expression contexts are deliberately not rebuilt: they are per-scanner and format-independent, and _init_expr_ctxes is not idempotent. Verified by disabling the rebuild and rerunning the suites: only the union read fails, with exactly the original error, and the four fluss suites that do not mix formats stay green. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
Every guard in the metadata gateway tells a lake handle from a fluss one by asking the sibling whether the handle is its own. Connector.ownsHandle defaults to false, so a sibling that never overrode it answers "not mine" about handles it has just produced: every guard falls through, and the first cast throws a ClassCastException naming two connectors and no cause. Checked once now, where the handle is born, in both places one is obtained — the $lake system-table handle and the union read's pinned lake handle. The message names the sibling's own class, so the fix is on the reader's screen instead of several layers away. The test double gains a switch for a sibling that inherits the default, because that is the shape of a real connector before it is first used as a sibling — and the only shape a double cannot have by accident. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
… log The fluss regression cluster gains lakehouse storage: a paimon warehouse on a bind mount, the tiering service submitted as a Flink job while the fixtures are built, and five lake tables. Building them takes three steps, and both of the unobvious ones are load-bearing. The rows that belong in paimon are written first; the paimon row COUNTS are then polled until they match, rather than waiting for a snapshot to exist, because tiering commits what it has consumed so far and a fixture frozen half-tiered would leave a lake-only table with a log tail on some runs and not others. Only then is the tiering job cancelled, and only then is the log tail written — left running, the service would keep consuming the tail, and a suite asserting that a table reads as "lake plus log" would decay into one asserting "lake only" as the environment aged. test_fluss_lake_only reads the lake through the sibling: its rows, the three system columns fluss adds, and the type identity between a table and its $lake — required to hold, and until now checked only by reading both mappings side by side rather than against the paimon connector running. test_fluss_union_log leans on comparing the two read modes. required reads lake plus log; disabled replays the whole fluss log, which still holds everything because tiering copies rather than moves. Two entirely different readers over one table, so a seam that is off by a row in either direction makes them disagree — an assertion a single-mode suite cannot make, and one the hand-written sibling in the unit tests cannot make either. Four environment details that each cost a full restart to find are recorded where they were needed: paimon builds a hadoop Configuration even for a directory warehouse; the fluss image chowns /opt/fluss to uid 9999 but never declares USER, so it runs as root and its paimon directories lock out the flink containers; tiering a primary-key table reads kv snapshot FILES, so the flink cluster needs remote.data.dir mounted; and a warehouse path with no scheme is read as HDFS, which fails at scan time rather than at catalog creation. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
A primary-key table that is tiered into a lake was refused outright, which left a plain SELECT on it failing. Merging its lake with its change log BY KEY is still not implemented -- the two cannot simply be concatenated, since the log carries updates and deletes of rows the lake already holds -- but refusing is the wrong answer, because reading such a table from fluss alone returns the WHOLE table: fluss keeps a primary-key table's state in its own kv store, and tiering copies rows into the lake rather than moving them out. That is the opposite of a log table, where whatever tiering has aged out of the log lives only in the lake and a fluss-only read silently loses it. So auto and disabled now read it the way an untiered primary-key table is read, from the latest kv snapshot plus the log that followed. What is lost is speed, not rows. required still refuses it: that mode exists so a union-read test cannot pass without a union read. The decision is taken BEFORE the lake snapshot is asked for, because it cannot depend on the answer -- asked afterwards, a table whose tiering has not committed yet would be refused under required with "wait for the tiering service to commit", which is a dead end rather than a reason. The old refusal also told the reader that switching the lake off returns less than the whole table. That is true of a log table and false here. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
The fixture's primary-key lake table now disagrees with its lake in all three ways a change log can: one tiered row is updated, another is deleted, and a key that the lake never saw is added. The last one is new here, and it is the case a merge of the two halves is most likely to get wrong -- the rows that exist only in the log have to be emitted exactly once no matter how many lake files the bucket has. The suite records what that table reads as today, next to what its lake holds, so the two can be seen not to be the same thing. Reading it from fluss alone is complete, so this baseline is also the answer a future lake+log merge has to reproduce row for row: when that lands, these recorded results must not change. Nothing in the result distinguishes a fluss-only read from a correct merge -- which is what makes it a baseline -- so the plan is asserted separately for which path actually ran. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
These suites asserted every result inline. That was a choice made before there was a cluster to run them on: a .out cannot be generated without one, and a hand-written .out is worse than none. There is a cluster now, so the results are recorded the way the rest of the regression tests record them. What is NOT a result stays in the code: the EXPLAIN anchors saying how a scan was planned, the refusal messages, the range-count bounds (which bucket a row lands in is fluss's choice, so they cannot be exact), and the two invariants that are comparisons rather than expectations -- that lake+log and fluss-only return the same rows, and that a table and its $lake report the same column types. Three values are deliberately left unrecorded because they are not reproducible: __bucket and __offset, which depend on the writer's bucket choice, and __timestamp, which is a wall clock. What is recorded of them is that every row has all three, within the range each must be in. The session time zone is now pinned, because TIMESTAMP_LTZ renders through it and the baseline holds what it rendered as. Recording the types tables whole also pins how a map, a struct and a decimal print, which the predicate-based assertions deliberately avoided. That is the trade: a rendering change now lands here and has to be re-recorded on purpose. In exchange the type coverage got stronger -- desc now pins the mapped types, and the same row read through fluss, through paimon and through a kv snapshot produces three baseline lines that are identical character for character. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
A scan range this connector plans is opaque about its origin: the JNI arm carries a serialized split and nothing else, the native arm a file path and a byte interval. That is fine while the only reader is BE, which just reads what it is handed. It stops being fine once another connector plans splits here on behalf of its own table. The fluss connector does exactly that: a fluss table tiered into paimon keeps a bucket-identical layout, and reading it means pairing the lake data of bucket b with the log tail of bucket b that has not been tiered yet. Nothing on the range says b. Parsing it out of the data-file path would work only on the native arm and only by depending on this connector's directory layout. So carry it: paimon.bucket = DataSplit.bucket(), on the native and JNI arms alike -- which BE reader a split lands on is a session-level escape hatch the sibling does not control, and it must not change what the sibling can learn. FE-only; populateRangeParams does not forward it, so BE sees nothing new. Two ranges deliberately do NOT carry it. The collapsed COUNT(*) range stands for the splits of every bucket, so any single number on it would be a lie. A non-DataSplit system split has no bucket at all. A consumer that needs the binding must fail loud on an absent bucket rather than read it as "no state for this bucket" -- that reading turns a broken contract into duplicated rows. The fixture is two-bucket on purpose: with one bucket every range reads "0" and a hard-coded constant passes. Four mutations checked red -- constant bucket on the native arm, no bucket on the JNI arm, a bucket on the system split, a bucket on the count range. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
Reading a tiered primary-key table means BE has to read the key columns of the log tail whether or not the query projects them -- they are what tells it which lake rows a newer log record has already replaced. That projection does not come from FE as slot descriptors the way every projection so far has; BE assembles it from names and types. Whether the reader accepts such a projection was an open question, and the answer decides the shape of the whole feature: if it does not, the key columns have to be driven straight into the JNI bridge as parameter strings instead. It does. The reader reads nothing off the columns but their name, type and partition flag, and both strings the Java scanner projects by are derived from that -- so a projection with no query behind it drives it the same as a planned one. Pins the two derived strings and the block C++ receives the columns into, because nothing between here and a wrong-columns read would notice. Checked red by skipping the derivation. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
A connector whose BE-side reader merges or suppresses rows by key needs that key read whether or not the query selected it. Doris already keeps those columns for its own aggregate and merge-on-read unique-key tables -- preserveExtraStorageKeySlots, four lines above where the scan's slots are pruned -- for exactly that reason. A plugin connector had no way to say the same thing, and BE cannot read a column the plan never asked for. So ask it: getMustReadColumns, answered per scan, empty by default, so nothing changes for a connector that needs only what the query projects. The answer arrives during plan translation, after the scan node is initialized and before splits are planned, and widens the scan's tuple only -- the project above it already has its own output tuple, so the column is read and then dropped rather than returned. The question goes through the same memoized provider that will plan the splits, because the two have to come from one decision: a connector that answers "no extra columns" here and then plans a read that needs them leaves BE looking for a column that is not in the projection. A name that matches no slot fails the query and says which name, rather than being skipped -- skipping turns a disagreement about the table into silently wrong rows. Checked red by six mutations: dropping the branch, skipping unknown names, stopping after the first match, dropping the null answer guard, resolving a fresh provider to ask, and a non-empty SPI default. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
…tail
A tiered primary-key table was read from fluss alone: correct, because fluss
keeps such a table's state in full, but it left the lake's columnar files
unread. It is now planned as three parts, per bucket:
- the sibling's lake splits, wrapped with the offsets of the log tail that
supersedes part of them (BE drops the lake rows whose keys the tail names),
- one PK_TAIL range producing the tail's own surviving state, exactly once,
- PK_FULL for a bucket the lake has never seen, as before.
A split is bound to the tail of its OWN bucket, using the bucket the paimon
connector now reports on each range. Three ways that can go wrong all fail
loud, because each would silently duplicate rows: a split that names no bucket
(an older paimon plugin), a bucket this table does not have (the two are not
bucketed alike), a bucket the lake holds files for but fluss records no tiering
offset for (their metadata disagrees).
Two conditions make the union unsafe rather than merely awkward, and under auto
both give up the lake half instead of failing — the fluss-only read they fall
back to is the whole table:
- a key or partition column whose values may not compare or render the same
way on both sides (float, a timestamp Doris rounds, a non-string partition
column, all of which fluss itself allows). Settled from the schema alone,
before anything is asked of the lake, so that the answer is the same at plan
translation time as it is when the ranges are planned;
- a tail the log no longer holds. Fluss deletes old log segments on a timer
that does not wait for tiering, and a primary-key table cannot re-read its
log from the lake, so planning verifies the tail is still there.
required refuses both, as it must: that mode exists so a test can assert the
lake was actually read. EXPLAIN gains suppressedLakeSplits, pkTailRanges and a
degraded= reason, which is otherwise the only thing distinguishing "there is no
lake" from "there is one this query could not use".
The scan keeps the key columns in its tuple through the SPI added for it, so BE
reads them as ordinary projected columns; the projection above the scan drops
them again. Both questions are answered from one memoized resolution, which is
a correctness requirement rather than a saving.
Removes the UNION_PK range that an earlier design would have used.
182 tests, 0 skipped. Ten mutations were confirmed to fail the new assertions,
including binding splits by bucket while ignoring the partition, reading a
missing bucket property as "no tail", skipping the tail guard, and resolving the
union read twice instead of reusing the memo.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
…t ended in A primary-key table read as its lake plus its log tail needs the fluss half of that read: the change log after the offset the lake was tiered at, replayed by key. Returning that log as it stands returns every intermediate state, so a key written three times comes back three times. The three stopping rules of a bounded log range now live in one place (BoundedLogRecords) instead of two: the log reader drops the change type and takes the row, the tail reader replays the change types by key. The rule they exist for -- a fetch that consumed up to the stop without yielding a record there -- only ever shows itself as a query that never returns, which is exactly the kind of thing a second copy gets wrong. What the tail owes is narrower than a whole-bucket read, and every test turns on it: a key nobody touched in the tail must NOT come back (the lake has it), a key updated must come back once with the later value, a key deleted must come back not at all. A delete leaves a tombstone rather than removing the key -- BE's C++ side hides the lake rows for every key this range touched, deleted ones included, so the tombstone is what makes "this row disappeared on purpose" countable instead of indistinguishable from a key never written. Replay needs the primary key whether or not the query selects it, so the reader appends the missing key columns to what it asks fluss for and keeps the requested columns at the front; the row loop reads those positionally and never learns anything was added. The tail is held in memory in a process shared by every query on this BE, so fluss.union_read.max_tail_rows bounds it and a tail that outgrew it is refused rather than read. 12 tests against a real cluster, 0 skipped; the 24 existing log and primary-key tests stay green through the refactor. Mutation run 7 of 8 red: ignoring deletes, removing the key instead of tombstoning it, reading from the start of the log, never reaching the ceiling, dropping the primary-key guard, accepting an empty range, and putting the key columns anywhere but the front. The eighth -- admitting a record AT the stopping offset -- survives because it is unreachable over a contiguous log: the rule below it stops at stop-1 first, so only a gap (control records) can reach it and a client cannot produce one. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
…tail replaced Planning already produces the lake half of a tiered primary-key table wrapped with the offsets of the log tail that supersedes part of it. This is the reader that honours that: it reads the tail's keys once per bucket and drops the lake rows those keys name. What the tail ended up saying about each key arrives separately, as its own range, so every row is produced exactly once and a key the tail ended by deleting is produced not at all. The lake half is read entirely by the paimon sibling's own reader stack -- every split is forwarded to it untouched, so native ORC/Parquet, serialized JNI splits, deletion vectors, schema evolution and split scheduling all stay its answer. This reader only removes rows. The filter runs on the block that comes back, in table-schema terms, rather than inside the sibling's file scan request. The key columns are ordinary projected columns there (planning keeps them in the tuple for this read), already resolved by the column mapper whichever way the lake table needs -- by name or by field id. Pushing the predicate down into the file request would save decoding the non-key columns of the suppressed rows, which is bounded by the size of the tail, and would cost a second place that resolves key columns against a file schema plus a separate implementation for the JNI child. The tail is read as a plain bounded log range, not as its surviving state: a lake row whose key the tail deleted has to disappear just as surely as one it updated, and the surviving state would not name it. One bucket's tail is one read however many of its splits this BE was given, held by the scan node's split cache under the tail's own offsets, so an unpartitioned table's bucket cannot collide with a partition's. Its size is bounded by the same limit and counted the same way as the Java half counts it -- change log records over the same offsets -- so a table over the limit is over it on both sides rather than in whichever half happened to run first. Three things fail loud instead of degrading, because each returns the superseded rows a second time and no row count reveals it: a key column the scan does not project (planning promised it would), a wrapped split with no tail bound to it, and a block read before a tail was bound. Aggregate pushdown is withheld from the lake half for the same reason: a COUNT answered from paimon's file metadata would count the rows this reader exists to remove, without ever producing a block to remove them from. 22 tests, 0 skipped; the 226 paimon, iceberg and table-reader tests around them stay green. Twelve mutations were confirmed red, including suppressing nothing, waving through a missing suppression, one cache entry for every bucket, a limit that never fires, an empty tail range, a skipped key column, forwarded aggregate pushdown, reading the tail as its surviving state, and reading the tail of a pruned split. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
A fluss primary-key table read as its lake plus its log tail plans its lake half as the paimon sibling's own split with a suppression descriptor attached, under its own table format name. Without a branch for that name the scanner refuses it outright, and the half of the read that the descriptor exists for never happens. Whether such a split can be read at all is the paimon answer, asked of the paimon payload it carries, rather than a second rule that could start disagreeing with paimon's after the sibling changes how it plans a split. That also keeps the C++ paimon reader's splits on the V1 fallback, where there is no branch for this format -- a clean refusal rather than a lake half read without its suppression. The reader-follows-range logic already covers a scan node holding all three kinds of range at once, which this one does. Checked red by removing the format from the accepted set. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
…hides test_fluss_lake_pk already recorded what such a table reads as, off a read that went to fluss alone -- an answer the merge of lake and log tail has to reproduce row for row. It does: that recorded file is unchanged here, which is the whole point of having recorded it before the merge existed. What a single-bucket fixture cannot show is WHERE each half comes from, since merging per bucket and merging per table are the same arrangement when there is one bucket. Three fixtures separate them: lake_pk_multi spreads nine keys over three buckets and gives only some of them a tail, lake_pk_part stands its three partitions differently towards the lake -- one lake plus tail, one lake alone, one written after tiering stopped and therefore read whole from fluss inside the same scan -- and lake_pk_cold keeps nothing in the log at all. Binding a tail to the wrong bucket suppresses nothing, because a key lives in exactly one bucket, and the rows that tail was meant to replace come back beside their replacements; of the suites here only the multi-bucket one notices. There is deliberately no deletion-vector fixture. Fluss does forward a 'paimon.deletion-vectors.enabled' table property into the paimon table it creates, which was verified against this environment, but its tiering service then writes data files and no index -- and paimon reads such a table as empty. A fixture that reads as empty asserts nothing. Two suites change with the code that landed before this one. `required` no longer refuses a primary-key table, so test_fluss_lake_pk asserts the merge and compares all three modes instead of expecting a refusal, and the refusal block in test_fluss_union_log becomes an assertion that the two table kinds still plan differently. Retrying the fixture build is fixed as well: init.sql drops its fluss database but leaves the paimon tables behind, and fluss refuses to create a lake table over a paimon table that already holds rows, so any attempt that failed halfway used to doom every attempt after it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
A fluss partition carries its value nowhere but in its own name, and fluss allows only ASCII letters, digits, '_' and '-' there -- so a value holding anything else is rewritten on the way in. A FLOAT 1.5 is named 1_5; a TIMESTAMP 2026-01-01 01:02:03 is named 2026-01-01-01-02-03. The substitution is many-to-one and Doris is handed the name, not the value, so nothing reads those back. Fluss creates such a table without complaint and DESC looks ordinary, so until now the first sign of trouble came from fe-core's partition parser: "failed to convert partition [1_5] to list partition" -- naming neither the column, nor its type, nor fluss, nor what to do instead. Decide it from the table's schema instead, before the partitions are listed, so the answer is the same whether the table has any yet or not, and say which column and which type. CHAR, STRING, BOOLEAN, the integer family and DATE are kept verbatim and pass; BINARY and BYTES are named with the hex text of their bytes, which reads back as the text it is unless enable.mapping.varbinary asks for a VARBINARY column, so their verdict is the one the catalog decides. Every verdict was established against a fluss cluster rather than reasoned from the naming rules, and the switch is exhaustive over fluss's type roots: a type a future release adds to its partition-key whitelist is refused with a message rather than waved through into the parser. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
…usand rows The seven suites so far each follow one path in depth, over fixtures small enough that a wrong answer is visible by reading it. Five things that leaves uncovered, and twelve fixtures for them: - Nesting. The type fixtures carry one column per fluss type and never more than one level, but an element decoder is chosen per level: being right about MAP<STRING,INT> says nothing about the MAP<STRING,ROW<..>> beside it. log_nested, pk_nested and lake_nested carry every combination of the three constructors, three levels deep, through all three row formats -- arrow, compacted and parquet-through-paimon -- with rows that are NULL at the outer level and rows whose collections hold NULL elements, which is a different thing to get wrong. - Partition column types. Everything so far partitions by STRING, which is also the one type that cannot tell a rendering bug from a working one. part_types partitions by one column of every type whose value survives fluss's partition naming and prunes on each of them; part_ts is the counter-case, and lake_pk_part_int is the primary-key table whose halves must not be matched by a non-STRING partition value -- covering the fallback that until now only ever fired on tables with no lake at all. - Size. big_log and big_pk hold 100000 rows each, tiered, with a tail. A scan that loses its last partial batch, a stopping offset applied to one bucket, a suppression set built per split rather than per bucket: none of those can show up on nine rows, where wrong and right are the same arrangement. Each is read through a union-read catalog and a disabled one, so one fixture answers the log, primary-key and union-read cases and the two paths are compared against each other. Every column is derived from a sequence, so the aggregates are closed forms rather than transcriptions. - Compound predicates. A disjunction over a partition column has to keep every partition it names; a predicate that reached only one half of a union read returns an answer short by exactly the other half's matching rows, and a plausible one. Three-valued logic gets its own block. - The rest of what a catalog answers for: tables with nothing in them (pk_empty, lake_empty, and the aggregates that are 0 and NULL at once), the fluss TIME column Doris has nowhere to put, the ways a query is supposed to fail, and the ordinary SQL a user reaches for -- joins, subqueries, set operations, CTAS and insert-select into an internal table. Every recorded block was checked against the fixture's own literals, and each new baseline was perturbed by one value to confirm it fails when it should. The existing baselines change only by the new table names in two listings; no recorded data row moved. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VCPjzhwMQuP7nTgdvGVbM
morningman
requested review from
924060929,
CalvinKirs,
Gabriel39,
englefly,
morrySnow,
starocean999 and
yiguolei
as code owners
August 3, 2026 23:11
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
4 tasks
Gabriel39
requested changes
Aug 4, 2026
Gabriel39
left a comment
Gabriel39
left a comment
Contributor
There was a problem hiding this comment.
I found two current CI blockers and one scan-wide memory-bound issue in the new union-read path. Details are inline.
| # The same library also duplicates zstd, lz4, snappy, bzip2 and zlib symbols. Those are C | ||
| # ABIs, stable across versions and layout-free, so they are left alone until something | ||
| # shows otherwise -- unlike RocksDB, whose C++ objects are what actually corrupt. | ||
| target_link_options(doris_be PRIVATE "-Wl,--exclude-libs,librocksdb.a") |
Contributor
There was a problem hiding this comment.
[P1] Please make this linker option platform-specific. --exclude-libs is a GNU/ELF linker option; Apple ld rejects it with ld: unknown options: --exclude-libs, which is exactly why the current BE UT (macOS) check fails while linking doris_be. Gate this option to the supported ELF platforms (and use a Darwin-specific solution only if the symbol-hiding behavior is needed there) so macOS can still build.
| @@ -0,0 +1 @@ | |||
| org.apache.doris.connector.fluss.FlussConnectorProvider | |||
Contributor
There was a problem hiding this comment.
[P1] Please add the ASF license header using # comments, as the existing connector service descriptors do. This new one-line descriptor currently makes License Check fail with an invalid/missing license header, so the PR cannot merge as-is.
| const auto cache_key = fmt::format("fluss_union_tail:{}:{}", tail.spec.size(), tail.spec); | ||
| Status read_status = Status::OK(); | ||
| bool cache_hit = false; | ||
| auto* cached = options.cache->get<SuppressionKeys>( |
Contributor
There was a problem hiding this comment.
[P1] Please bound or release these cached tail-key blocks at the scan level. fluss.union_read.max_tail_rows limits one partition/bucket tail, but ShardedKVCache has no eviction and lives for the entire FileScanLocalState, so every tail touched by every partition and bucket remains resident until the scan finishes. A partitioned table can therefore retain partition_count * bucket_count * max_tail_rows key records (the default is two million per tail), even after all lake splits for earlier tails have completed, and hit the query/BE memory limit although the input can otherwise be streamed. Consider a scan-wide byte/row budget plus eviction/ref-counting after the last split using a tail, and deduplicate touched keys before retaining them.
Gabriel39
reviewed
Aug 4, 2026
| return table_format == "NotSet" || table_format == "tvf" || table_format == "hive" || | ||
| table_format == "iceberg" || table_format == "paimon" || table_format == "hudi"; | ||
| table_format == "iceberg" || table_format == "paimon" || table_format == "hudi" || | ||
| // A lake split of a fluss primary-key table read as its lake plus its log tail. It is the |
Contributor
There was a problem hiding this comment.
[P2] Please handle the legacy-scanner selection explicitly. All Fluss reader dispatch added by this PR exists only in FileScannerV2, but FileScanLocalState::_should_use_file_scanner_v2 still honors enable_file_scanner_v2=false; in that supported session configuration the scan falls into FileScanner, whose FORMAT_JNI dispatch has no fluss/fluss_union branch and returns Not supported create reader for table format. Either force V2 for Fluss ranges (if the variable is only a preference for scans supported by both implementations), or fail during planning with a clear requirement; add a test with the variable disabled so this does not remain a late BE runtime surprise.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What problem does this PR solve?
Issue Number: close #xxx
Related PR: #66403
Problem Summary:
Doris cannot read Apache Fluss (incubating) tables. This adds a fluss catalog, reading all three shapes a fluss table can take:
table.datalake.enabled)For a tiered table the lake half is not re-implemented: it is planned by the paimon connector, on the snapshot the fluss coordinator pinned, so that half gets native ORC/Parquet, deletion vectors and the file cache for free — and the fluss plugin ships no paimon dependency at all (it borrows the paimon plugin through
createSiblingConnector, the way hive borrows iceberg and hudi). A tiered table also exposestbl$lake, which reads the lake alone.For a tiered primary-key table the two halves are merged BY KEY, natively in BE: FE wraps each lake split with the key set of that bucket's log tail, the BE reader drops the lake rows that tail replaced before materialization, and the tail is replayed once as its own range.
CREATE CATALOG fluss PROPERTIES ( "type" = "fluss", "fluss.bootstrap.servers" = "host:9123" ); SELECT * FROM fluss.db.tbl; -- lake + log, merged SELECT * FROM fluss.db.`tbl$lake`; -- the lake aloneReviewing this
35 commits. Five of them are not in the fluss module and stand on their own, so they are proposed
separately in #66403 and are not the subject of this review. They are still part of this branch until
that lands; once it does, this PR is rebased on top of it and shrinks by exactly those five commits.
Listed here so a reviewer of this branch knows what they are and where they went:
[fix](be) Stop exporting the statically linked RocksDB symbolsbe/src/service/CMakeLists.txtdoris_beexports 4840 statically linked rocksdb symbols, so any JNI library carrying its own RocksDB resolves 2576 of them into doris_be's copy — across two different libstdc++ string ABIs. An object built with one layout, used by functions compiled for another ⇒bad_allocthrough a JNI frame ⇒ the BE process aborts. Fluss's embedded frocksdbjni is simply the first library to hit it.[fix](be) Pick the table reader per scan range, not per scan nodebe/src/exec/scan/file_scanner_v2.{h,cpp}_open_implbuilds one_table_readerfrom the first split and every later split reuses it. A scan node holding twotable_format_types — the shape any union read has — hands the second kind to the first kind's reader. The symptom is non-deterministic, because which splits share a scanner is the engine's choice.[fix](paimon) Claim the table handles this connector producesfe/fe-connector/fe-connector-paimonConnector.ownsHandledefaults to false; iceberg and hudi override it, paimon never did. Any connector using paimon as a sibling is told "not mine" about every handle paimon itself produced.[feat](paimon) Say which bucket a scan range came fromfe/fe-connector/fe-connector-paimonLakeSplit.bucket()as first class; this is the same fact on the Doris side.[feat](connector) Let a connector name the columns its reader must readfe/fe-connector/fe-connector-api+fe/fe-corepreserveExtraStorageKeySlots+extra_key_column_slot_ids), for the same reason. The new branch sits beside that one, before the sameremoveIf. The SPI method defaults to an empty set, so every other connector is unaffected.The remaining 30 commits are the connector itself, grouped so that each group compiles, runs and has its own acceptance:
$lakethrough the paimon sibling, and lake-plus-log for log tablesThings worth saying out loud
fluss 1.0-SNAPSHOT, resolved from the Apache snapshot repositoryfe/pom.xmlalready declares. Fluss 1.0 is not released and the APIs this uses are not in 0.9.1. The coordinates switch to the release when there is one. If CI goes red while this is green locally, compare the fluss snapshot timestamps first.enableFlussTestdefaults to false and the external pipeline's conf deliberately does not set it. That switches on with the fluss release.TIMESTAMPpushed into paimon matches no row (a plain paimon catalog over the same warehouse behaves the same), andjava-common's VARBINARY read-back decodes what it wrote as a StringView, so it always reads zeros (the write side and BE agree; only the Java read-back is wrong, and no production path takes it).TIMEis mapped toUNSUPPORTEDrather than to a string or to elapsed millis: Doris has no type meaning what fluss's TIME means, and handing back a plausible value of a different meaning is not something a later error would catch. The paimon and iceberg connectors mark their own TIME the same way, which also keepstblandtbl$lakeagreeing on one schema.filesystemis supported as the paimon lake catalog for now. HMS and REST lake catalogs, predicate pass-through into the lake half, and the write path are follow-ups.Release note
Support reading Apache Fluss tables through a new
flusscatalog, including tables tiered into a paimon lake, which are read as the lake plus the change log written after it.Check List (For Author)
Test
regression-test/suites/external_table_p0/fluss— 12 suites, 0 skipped, green on two consecutive runs against a real fluss + flink + paimon cluster (docker/thirdparties/docker-compose/fluss): catalog, log tables, primary-key tables, lake-only, lake+log, primary-key merge, nested complex types, partition-column types, 100k rows, compound predicates, and empty/negative/misc.Unit tests:
fe-connector-fluss192,fe-connector-paimon511,fluss-scanner36,fe-connector-api+fe-core(the must-read-columns chain) 36 — all 0 skipped. BE:FileScannerV2*:FileScannerTest*:Fluss*55 andPaimon*:*Iceberg*:*EqualityDelete*227, all passing.Every new behaviour was mutation-tested: the change was inverted and the suite had to go red. That includes the end-to-end ones — the merge was mutated in C++, in BE-java and in FE, rebuilt and redeployed each time, and each mutation turned a suite red.
Behavior changed:
Does this need documentation?