[poc] FIP-45 Log Enrichment

FIP45-POC.md

@@ -0,0 +1,78 @@
+# FIP-45 proof of concept: column groups as shadow logs
+
+Branch `fip45-v2`, based on `apache/fluss` main at `da69aee23` (2026-09-03).
+
+This branch implements the revised FIP-45 design that answers the dev@ review of the first
+proposal (see `FIP45-PLAN.md`). It covers work packages
+WP1 (storage) and WP3 (read path and `listOffsets`) of that plan end to end, plus the minimum of
+WP7 (client write API) needed to drive them, and it ships an integration test proving the
+contract.
+
+## What changed, in one paragraph
+
+A column group is a **shadow log**: a chain of ordinary `LogSegment`s stored under
+`{tabletDir}/col-{group}/`, whose batches hold only the group's columns and whose batch base
+offsets are the base-log offsets they fill. The group's log end offset is the enrichment
+watermark and its high watermark is the committed enrichment watermark, so the base log's index,
+recovery, truncation and zero-copy read code apply unchanged. The base log of a column-group
+table physically stores only the default-group columns. On a fetch, the server derives the touched
+groups from the projection, clamps the read at `min(HW, CEW_g)`, serves the base batches as file
+slices exactly as today (unprojected or through `FileLogProjection`), and ships each touched
+group's batches for the same offset range as additional zero-copy payload. The client
+merge-joins base and group rows by offset when it materialises `ScanRecord`s, so batch metadata
+(`commitTimestamp`, `baseLogOffset`, `writerId`, statistics) is never rewritten.
+
+## Files
+
+| Area | Files |
+|---|---|
+| Schema | `Schema` (column groups, base/group row types, name validation), `ColumnJsonSerde`, `TableDescriptor`, `TablePath`, `ColumnGroupSchemaGetter` |
+| Protocol | `FlussApi.proto`: `ProduceLogColumns*`, `PbColumnGroupRecords` on fetch responses, `PbColumnGroupFetch` on fetch requests, `column_group` on `ListOffsetsRequest`; `ApiKeys.PRODUCE_LOG_COLUMNS`; `Errors` 74 to 77; `TabletServerGateway.produceLogColumns` |
+| Server storage | `ColumnGroupLog`, `ColumnGroupAppendInfo`, `LogTablet` (load, append validation, ranged read, bounded read, truncation), `LocalLog.convertToBatchEndOffsetMetadata`, `FlussPaths.columnGroupLogDir` |
+| Server read/write | `Replica` (gate, group payload, group high watermark, `appendColumnsAsLeader`, group `listOffsets`), `ReplicaManager.appendColumnsToLog`, `ColumnGroupFetchPlan`, `FetchParams`, `LogReadInfo`, `ServerRpcMessageUtils`, `TabletService.produceLogColumns`, `FileLogProjection.lastProjectedOffset` |
+| Client write | `AppendWriter.appendColumns`, `AppendWriterImpl` (base-only physical row), `ColumnGroupWriter`, `WriterClient`, `RecordAccumulator` (base row type) |
+| Client read | `ColumnGroupReadPlan`, `ColumnGroupStitcher`, `CompletedFetch`, `DefaultCompletedFetch`, `LogFetcher`, `LogRecordReadContext` (physical row type factory) |
+| Tests | `ColumnGroupLogTest` (server), `ColumnGroupITCase` (client), plus the schema tests from the first POC |
+
+## Contract implemented
+
+- `appendColumns(group, bucket, firstSourceOffset, rows)`: `firstSourceOffset` must equal the
+  group's log end offset; a batch entirely below it is acknowledged as a duplicate; a batch that
+  straddles it or leaves a gap fails with `InvalidColumnGroupOffsetException` carrying the expected
+  offset; the last row must be below the base high watermark.
+- Offsets that base retention removed are trivially complete: the group log advances to the base
+  log start offset before validation.
+- Reads whose projection touches no group read to HW, byte for byte as before. Reads touching
+  groups are clamped at the smallest committed enrichment watermark among them. Because batches are
+  never split, the base read may include the batch containing the watermark; group rows are shipped
+  only up to the watermark, and the client stops exactly there and fetches again from it.
+- A projection with only group columns carries the first base column so the fetch advances.
+- `listOffsets(LATEST)` returns the base high watermark for every caller. With `column_group` set
+  it returns the group's high watermark; `LEADER_END_OFFSET_SNAPSHOT` returns the group's log end
+  offset.
+
+## Running the tests
+
+```bash
+./mvnw -o install -DskipTests -pl fluss-common,fluss-rpc,fluss-server,fluss-client
+./mvnw -o test -pl fluss-server -Dtest=ColumnGroupLogTest
+./mvnw -o verify -pl fluss-client -Dtest=ColumnGroupITCase -Dit.test=ColumnGroupITCase \
+    -DfailIfNoTests=false -Dsurefire.failIfNoSpecifiedTests=false
+```
+
+## Not in this proof of concept
+
+- **Follower replication of column groups (WP2).** The fetch protocol carries the follower cursor
+  field and the leader tracks reported follower end offsets, but `ReplicaFetcherThread` does not
+  send cursors or append shipped group records yet, and the group high watermark is not
+  checkpointed. A follower that has not reported is not counted in the group high watermark, so
+  with replication factor greater than one the committed enrichment watermark currently equals the
+  leader's enrichment watermark. The integration tests use replication factor 1.
+- **Remote tiering of group segments and remote reads with companions (WP4)**, lake tiering
+  changes (WP5) and the Flink connector (WP6).
+- **Client batching, one-in-flight ordering, leader-change retries and resync on
+  `expected_source_offset` (WP7).** Every `appendColumns` call is one request.
+- **`acks`** on `ProduceLogColumns` is accepted but the response is sent after the local append.
+- Projection of a subset of a group's columns on the server (all group columns are shipped; the
+  client selects), Arrow-batch polling on projections touching groups, group segment retention,
+  and metrics.

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ColumnGroupReadPlan.java

@@ -0,0 +1,220 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table.scanner.log;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.metadata.ColumnGroupSchemaGetter;
+import org.apache.fluss.metadata.LogFormat;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.SchemaGetter;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.record.LogRecordReadContext;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.AllocationManager;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.utils.Projection;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * How a scan of a column-group table (FIP-45) maps its output columns onto the physical logs: the
+ * base log holds the default-group columns, each column group holds its own columns, and the client
+ * stitches rows from them by offset.
+ */
+@Internal
+public final class ColumnGroupReadPlan {
+
+    /** Projection over the base physical row, in output order; null for the whole base row. */
+    @Nullable private final Projection baseProjection;
+
+    /** Column groups touched by the scan, in order of first appearance. */
+    private final List<String> touchedGroups;
+
+    private final Map<String, RowType> groupRowTypes;
+
+    /** Per output column: -1 for the base log, else the index into {@link #touchedGroups}. */
+    private final int[] outputSource;
+
+    /** Per output column: index into the base output row or into the group's physical row. */
+    private final int[] outputField;
+
+    private ColumnGroupReadPlan(
+            @Nullable Projection baseProjection,
+            List<String> touchedGroups,
+            Map<String, RowType> groupRowTypes,
+            int[] outputSource,
+            int[] outputField) {
+        this.baseProjection = baseProjection;
+        this.touchedGroups = touchedGroups;
+        this.groupRowTypes = groupRowTypes;
+        this.outputSource = outputSource;
+        this.outputField = outputField;
+    }
+
+    /** Plans a scan of {@code schema} with {@code projection}; null if the table has no groups. */
+    @Nullable
+    public static ColumnGroupReadPlan of(Schema schema, @Nullable Projection projection) {
+        if (!schema.hasColumnGroups()) {
+            return null;
+        }
+        int[] baseIndices = schema.getDefaultGroupColumnIndices();
+        Map<String, List<Integer>> groups = schema.getColumnGroups();
+        int[] outputColumns =
+                projection == null
+                        ? allColumns(schema.getColumns().size())
+                        : projection.getProjection();
+
+        List<String> touched = new ArrayList<>();
+        List<Integer> baseFields = new ArrayList<>();
+        int[] outputSource = new int[outputColumns.length];
+        int[] outputField = new int[outputColumns.length];
+        for (int i = 0; i < outputColumns.length; i++) {
+            int column = outputColumns[i];
+            String group = schema.getColumnGroupOf(column);
+            if (group == null) {
+                int basePosition = indexOf(baseIndices, column);
+                outputSource[i] = -1;
+                if (projection == null) {
+                    outputField[i] = basePosition;
+                } else {
+                    outputField[i] = baseFields.size();
+                    baseFields.add(basePosition);
+                }
+            } else {
+                int groupIndex = touched.indexOf(group);
+                if (groupIndex < 0) {
+                    groupIndex = touched.size();
+                    touched.add(group);
+                }
+                outputSource[i] = groupIndex;
+                outputField[i] = groups.get(group).indexOf(column);
+            }
+        }
+        Projection baseProjection = null;
+        if (projection != null) {
+            if (baseFields.isEmpty()) {
+                // the server cannot project zero columns: carry the first base column, which the
+                // output mapping never references
+                baseFields.add(0);
+            }
+            baseProjection =
+                    Projection.of(baseFields.stream().mapToInt(Integer::intValue).toArray());
+        }
+        Map<String, RowType> groupRowTypes = new HashMap<>();
+        for (String group : touched) {
+            groupRowTypes.put(group, schema.getColumnGroupRowType(group));
+        }
+        return new ColumnGroupReadPlan(
+                baseProjection,
+                Collections.unmodifiableList(touched),
+                groupRowTypes,
+                outputSource,
+                outputField);
+    }
+
+    private static int[] allColumns(int count) {
+        int[] all = new int[count];
+        for (int i = 0; i < count; i++) {
+            all[i] = i;
+        }
+        return all;
+    }
+
+    private static int indexOf(int[] array, int value) {
+        for (int i = 0; i < array.length; i++) {
+            if (array[i] == value) {
+                return i;
+            }
+        }
+        throw new IllegalArgumentException("Column " + value + " is not a base column.");
+    }
+
+    public List<String> touchedGroups() {
+        return touchedGroups;
+    }
+
+    public int outputCount() {
+        return outputSource.length;
+    }
+
+    int outputSource(int output) {
+        return outputSource[output];
+    }
+
+    int outputField(int output) {
+        return outputField[output];
+    }
+
+    public RowType groupRowType(String group) {
+        return groupRowTypes.get(group);
+    }
+
+    /** The read context for the base log, decoding only the default-group columns. */
+    public LogRecordReadContext createBaseContext(
+            TableInfo tableInfo,
+            boolean readFromRemote,
+            LogRecordReadContext.SchemaResolution schemaResolution,
+            SchemaGetter schemaGetter,
+            AllocationManager.Factory allocationManagerFactory) {
+        return LogRecordReadContext.createReadContext(
+                tableInfo.getTableId(),
+                LogFormat.ARROW,
+                tableInfo.getSchemaId(),
+                tableInfo.getSchema().getBaseRowType(),
+                readFromRemote,
+                schemaResolution,
+                baseProjection,
+                ColumnGroupSchemaGetter.base(schemaGetter),
+                allocationManagerFactory);
+    }
+
+    /** One read context per touched column group, decoding that group's physical row. */
+    public Map<String, LogRecordReadContext> createGroupContexts(
+            TableInfo tableInfo,
+            SchemaGetter schemaGetter,
+            AllocationManager.Factory allocationManagerFactory) {
+        Map<String, LogRecordReadContext> contexts = new HashMap<>();
+        for (String group : touchedGroups) {
+            contexts.put(
+                    group,
+                    LogRecordReadContext.createReadContext(
+                            tableInfo.getTableId(),
+                            LogFormat.ARROW,
+                            tableInfo.getSchemaId(),
+                            groupRowTypes.get(group),
+                            false,
+                            LogRecordReadContext.SchemaResolution.TARGET,
+                            null,
+                            ColumnGroupSchemaGetter.group(schemaGetter, group),
+                            allocationManagerFactory));
+        }
+        return contexts;
+    }
+
+    /** Field getters over the physical row of {@code group}. */
+    public InternalRow.FieldGetter[] groupFieldGetters(String group) {
+        return InternalRow.createFieldGetters(groupRowTypes.get(group));
+    }
+}