[spark] Expose written columns for streaming micro-batches

docs/docs/spark/structured-streaming.md

@@ -198,6 +198,39 @@ val query = spark.readStream
   .start()
 ```
 
+### Written Columns of a Micro-Batch
+
+`foreachBatch` consumers can inspect which Paimon field IDs were written by the data files admitted to the current micro-batch. This experimental metadata collection is disabled by default. Enable `read.stream.batch-written-columns.enabled` on the Paimon streaming source, then call `PaimonSparkMicroBatchMetadata.writtenColumns` with the raw `Dataset` passed to `foreachBatch`.
+
+```scala
+import org.apache.paimon.spark.PaimonSparkMicroBatchMetadata
+import org.apache.paimon.table.source.{AllColumns, KnownWrittenColumns}
+import org.apache.spark.sql.{Dataset, Row}
+
+val query = spark.readStream
+  .format("paimon")
+  .option("read.stream.batch-written-columns.enabled", "true")
+  .table("table_name")
+  .writeStream
+  .option("checkpointLocation", "/path/to/checkpoint")
+  .foreachBatch { (batch: Dataset[Row], _: Long) =>
+    val writtenColumns = PaimonSparkMicroBatchMetadata.writtenColumns(batch)
+    if (!writtenColumns.isPresent) {
+      // Metadata is unavailable; conservatively process all columns.
+    } else if (writtenColumns.get() == AllColumns.INSTANCE) {
+      // Exact field IDs are unavailable; conservatively process all columns.
+    } else {
+      val fieldIds = writtenColumns.get().asInstanceOf[KnownWrittenColumns].fieldIds()
+      // Process the exact set of written Paimon field IDs.
+    }
+  }
+  .start()
+```
+
+A present `KnownWrittenColumns` contains the complete, immutable set of written field IDs in ascending order. The set may be empty; that is a known empty set, not unknown metadata. A present `AllColumns.INSTANCE` means that exact file or schema metadata could not be resolved, so every column must be treated as written.
+
+An empty `Optional` means that metadata is unavailable, for example because collection was not enabled, the micro-batch is empty, the `Dataset` is not the raw batch from a query with exactly one distinct Paimon streaming source, or its lineage is incomplete or ambiguous. An empty `Optional` does not mean that no columns were written; callers must fall back to processing all columns.
+
 Paimon Structured Streaming supports read row in the form of changelog (add rowkind column in row to represent its
 change type) in two ways:
 

docs/generated/spark_connector_configuration.html

@@ -44,6 +44,12 @@
             <td>Boolean</td>
             <td>Whether to read row in the form of changelog (add rowkind column in row to represent its change type).</td>
         </tr>
+        <tr>
+            <td><h5>read.stream.batch-written-columns.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to expose the written field ids of an admitted micro-batch through PaimonSparkMicroBatchMetadata.</td>
+        </tr>
         <tr>
             <td><h5>read.stream.maxBytesPerTrigger</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-core/src/main/java/org/apache/paimon/table/source/AllColumns.java

@@ -0,0 +1,27 @@
+/*
+ * 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.paimon.table.source;
+
+import org.apache.paimon.annotation.Experimental;
+
+/** A conservative marker requiring consumers to assume that every column may be written. */
+@Experimental
+public enum AllColumns implements WrittenColumns {
+    INSTANCE
+}

paimon-core/src/main/java/org/apache/paimon/table/source/KnownWrittenColumns.java

@@ -0,0 +1,67 @@
+/*
+ * 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.paimon.table.source;
+
+import org.apache.paimon.annotation.Experimental;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.TreeSet;
+
+/** A complete, immutable set of written field ids, ordered by field id. */
+@Experimental
+public final class KnownWrittenColumns implements WrittenColumns {
+
+    private static final long serialVersionUID = 1L;
+
+    private final List<Integer> fieldIds;
+
+    public KnownWrittenColumns(Collection<Integer> fieldIds) {
+        this.fieldIds = Collections.unmodifiableList(new ArrayList<>(new TreeSet<>(fieldIds)));
+    }
+
+    public List<Integer> fieldIds() {
+        return fieldIds;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof KnownWrittenColumns)) {
+            return false;
+        }
+        KnownWrittenColumns that = (KnownWrittenColumns) o;
+        return fieldIds.equals(that.fieldIds);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(fieldIds);
+    }
+
+    @Override
+    public String toString() {
+        return "KnownWrittenColumns{" + "fieldIds=" + fieldIds + '}';
+    }
+}

paimon-core/src/main/java/org/apache/paimon/table/source/WrittenColumns.java

@@ -0,0 +1,33 @@
+/*
+ * 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.paimon.table.source;
+
+import org.apache.paimon.annotation.Experimental;
+
+import java.io.Serializable;
+
+/**
+ * Columns written by the data files selected for a scan.
+ *
+ * <p>The result is either {@link KnownWrittenColumns} or {@link AllColumns}. Consumers must treat
+ * {@link AllColumns} conservatively and must not interpret an empty {@link KnownWrittenColumns} as
+ * unknown.
+ */
+@Experimental
+public interface WrittenColumns extends Serializable {}

paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java

@@ -20,13 +20,21 @@
 
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.table.source.AllColumns;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.KnownWrittenColumns;
+import org.apache.paimon.table.source.WrittenColumns;
 import org.apache.paimon.types.DataField;
 
 import java.util.Collection;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -38,6 +46,72 @@
 /** Util class for data evolution. */
 public class DataEvolutionUtils {
 
+    /** Collect written field ids from data files in the selected splits. */
+    public static WrittenColumns collectWrittenColumns(
+            Collection<DataSplit> splits, Function<Long, TableSchema> schemaLoader) {
+        Set<Integer> fieldIds = new TreeSet<>();
+        Map<Long, Map<String, Integer>> fieldIdByNameCache = new HashMap<>();
+        Map<Pair<Long, List<String>>, Set<Integer>> fieldIdsCache = new HashMap<>();
+        for (DataSplit split : splits) {
+            for (DataFileMeta file : split.dataFiles()) {
+                try {
+                    Pair<Long, List<String>> cacheKey = Pair.of(file.schemaId(), file.writeCols());
+                    Set<Integer> fileFieldIds = fieldIdsCache.get(cacheKey);
+                    if (fileFieldIds == null) {
+                        fileFieldIds = computeFileFieldIds(schemaLoader, fieldIdByNameCache, file);
+                        fieldIdsCache.put(cacheKey, fileFieldIds);
+                        fieldIds.addAll(fileFieldIds);
+                    }
+                } catch (RuntimeException e) {
+                    return AllColumns.INSTANCE;
+                }
+            }
+        }
+        return new KnownWrittenColumns(fieldIds);
+    }
+
+    private static Set<Integer> computeFileFieldIds(
+            Function<Long, TableSchema> schemaLoader,
+            Map<Long, Map<String, Integer>> fieldIdByNameCache,
+            DataFileMeta file) {
+        Map<String, Integer> fieldIdByName =
+                fieldIdByNameCache.computeIfAbsent(
+                        file.schemaId(),
+                        schemaId -> {
+                            TableSchema fileSchema = schemaLoader.apply(schemaId);
+                            if (fileSchema == null) {
+                                throw new IllegalArgumentException(
+                                        "Cannot find schema " + schemaId);
+                            }
+
+                            Map<String, Integer> fieldIds = new HashMap<>();
+                            for (DataField field : fileSchema.fields()) {
+                                fieldIds.put(field.name(), field.id());
+                            }
+                            return fieldIds;
+                        });
+
+        List<String> writeCols = file.writeCols();
+        if (writeCols == null) {
+            return new TreeSet<>(fieldIdByName.values());
+        }
+
+        Set<Integer> fieldIds = new TreeSet<>();
+        for (String writeCol : writeCols) {
+            Integer fieldId = fieldIdByName.get(writeCol);
+            if (fieldId == null) {
+                checkArgument(
+                        SpecialFields.isSystemField(writeCol),
+                        "Cannot find write column '%s' in schema %s.",
+                        writeCol,
+                        file.schemaId());
+            } else {
+                fieldIds.add(fieldId);
+            }
+        }
+        return fieldIds;
+    }
+
     /**
      * Table field ids physically present in a file, resolved through the schema used to write it.
      */