JingsongLi commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4254417732
##########
paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java:
##########
@@ -387,8 +412,99 @@ private FileRecordReader<InternalRow> createFileReader(
}
if (deletionVector != null && !deletionVector.isEmpty()) {
- return new ApplyDeletionVectorReader(fileRecordReader,
deletionVector);
+ fileRecordReader = new ApplyDeletionVectorReader(fileRecordReader,
deletionVector);
}
- return fileRecordReader;
+ return applyMetadataFallbackAndOuterProjection(fileRecordReader);
+ }
+
+ private RowType readTypeWithMetadataDependencies(RowType
requestedReadType) {
+ if (changelogExtraValueFields.isEmpty()) {
+ return requestedReadType;
+ }
+
+ List<String> readFieldNames = requestedReadType.getFieldNames();
+ List<DataField> dependencies = new ArrayList<>();
+ for (String preserveColumn : metadataPreserveColumns) {
+ String metadataName = metadataFieldPrefix + preserveColumn;
+ if (readFieldNames.contains(metadataName)
+ && !readFieldNames.contains(preserveColumn)
+ && schema.logicalRowType().containsField(preserveColumn)) {
+
dependencies.add(schema.logicalRowType().getField(preserveColumn));
Review Comment:
[P2] Project nested fallback values to the requested metadata shape
Adding the whole source field here is unsafe when metadata is nested-pruned.
With `payload ROW<a INT,b INT>` and `event-metadata-fields=payload`, requesting
only `__internal__payload.b` gives a metadata type `ROW<b INT>`, but
`FallbackMappingRow.getRow` returns the full physical `ROW<a,b>` unchanged. A
real table updated to `(a=33,b=44)` returns **33 instead of 44** for that
metadata child. I reproduced this with Parquet and Avro on the raw path and
with the merge reader forced via `forceKeepDelete()`. Spark nested pruning
produces this requested type. `MergeFileSplitRead.adjustReadType` has the same
dependency logic. Please combine physical/metadata dependency projections and
project each fallback value to its requested nested shape, including when the
physical source and metadata request different children.
##########
paimon-flink/paimon-flink-common/pom.xml:
##########
@@ -190,6 +190,13 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
Review Comment:
[P2] Remove the duplicate Avro dependency
This new test dependency exactly duplicates the existing unconditional
`org.apache.avro:avro:${avro.version}` test dependency at lines 108–113. The
exact-head [Flink/Spark E2E
CI](https://github.com/apache/paimon/actions/runs/37482976432/job/112339853634)
stops before compilation/tests with `dependencies.dependency ... must be unique
... duplicate declaration ... @ line 193`. Please retain one declaration so the
E2E build can run.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java:
##########
@@ -173,25 +179,61 @@ public RowLevelModificationScanContext
applyRowLevelModificationScan(
}
public Map<String, DataType> listReadableMetadata() {
- // Flink calls this after applyRowLevelModificationScan for row-level
operations.
- if (rowLevelModificationSnapshotId == null || !isDataEvolutionTable())
{
- return Collections.emptyMap();
- }
Map<String, DataType> metadata = new LinkedHashMap<>();
- metadata.put(SpecialFields.ROW_ID.name(),
DataTypes.BIGINT().notNull());
+
+ // Row-level modification metadata
+ if (rowLevelModificationSnapshotId != null && isDataEvolutionTable()) {
+ metadata.put(SpecialFields.ROW_ID.name(),
DataTypes.BIGINT().notNull());
+ }
+
+ // Event metadata from changelog-producer.event-metadata-fields
+ List<String> preserveColumns = eventPreserveColumns();
+ if (!preserveColumns.isEmpty() && table instanceof FileStoreTable) {
+ TableSchema schema = ((FileStoreTable) table).schema();
+ CoreOptions coreOptions = CoreOptions.fromMap(table.options());
+ for (DataField field :
+ ChangelogEventMetadata.extraValueFields(
+ schema.logicalRowType(), schema.highestFieldId(),
coreOptions)) {
+ DataType flinkType =
+ TypeConversions.fromLogicalToDataType(
+
LogicalTypeConversion.toLogicalType(field.type()));
+ metadata.put(field.name(), flinkType.nullable());
+ }
+ }
+
return metadata;
}
public void applyReadableMetadata(List<String> metadataKeys, DataType
producedDataType) {
for (String metadataKey : metadataKeys) {
- if (!SpecialFields.ROW_ID.name().equals(metadataKey)) {
- throw new UnsupportedOperationException(
- "Unsupported Paimon metadata column: " + metadataKey);
+ if (SpecialFields.ROW_ID.name().equals(metadataKey)
+ || eventMetadataFieldNames().contains(metadataKey)) {
+ continue;
}
+ throw new UnsupportedOperationException(
+ "Unsupported Paimon metadata column: " + metadataKey);
}
this.metadataKeys = metadataKeys;
Review Comment:
[P2] Apply declared metadata to temporal lookup results too
The scan path consumes these metadata keys, but the inherited
`getLookupRuntimeProvider` still passes the physical `FileStoreTable` and
physical projection to `FileStoreLookupFunction`. Flink accepts metadata on a
lookup table and expects the produced metadata column in each returned row.
Reproduction: declare `(id INT PRIMARY KEY NOT ENFORCED, event_ts BIGINT,
writetime BIGINT METADATA FROM '__internal__event_ts' VIRTUAL)`, preserve
`event_ts`, insert `(1,50)`, and run a processing-time temporal join selecting
`d.event_ts,d.writetime`. The actual job fails with
`ArrayIndexOutOfBoundsException: 2` in `GenericRow.isNullAt` through the
generated lookup collector because only the two physical fields are returned.
The same join selecting only `d.event_ts` passes. Please supply the requested
metadata in the lookup runtime (or explicitly reject the unsupported lookup
plan before launching a job) and add this integration case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
