fix(plugin-iceberg, native): Fix wrong results when reading row lineage columns

presto-docs/src/main/sphinx/connector/iceberg.rst

@@ -942,10 +942,13 @@ See the `Iceberg spec — Row Lineage <https://iceberg.apache.org/spec/#row-line
 ``_row_id`` column
 ~~~~~~~~~~~~~~~~~~
 A globally unique, monotonically increasing ``BIGINT`` identifier for each row.
-The value is derived from the data file's ``firstRowId`` (stored in the Iceberg
-manifest) plus the row's ordinal position within that file
-(``firstRowId + position_in_file``).  Row IDs are stable across reads and are
-unique within the table at any given point in time.
+If the data file stores a ``_row_id`` for the row, that value is returned.
+Otherwise the value is derived from the data file's ``firstRowId`` (stored in
+the Iceberg manifest) plus the row's ordinal position within that file
+(``firstRowId + position_in_file``).  Engines that preserve row lineage store
+each row's original ``_row_id`` when they rewrite the row, for example during
+compaction or a row-preserving ``UPDATE`` or ``MERGE``.  Row IDs are stable
+across reads and are unique within the table at any given point in time.
 
 .. code-block:: sql
 
@@ -977,7 +980,11 @@ The Iceberg data sequence number of the snapshot in which this row was last
 written (inserted or updated).  All rows written in the same ``APPEND`` or
 ``OVERWRITE`` commit share the same sequence number.  Later commits produce
 higher sequence numbers, so you can use this column to determine the relative
-write order of rows across snapshots.
+write order of rows across snapshots.  If the data file stores a
+``_last_updated_sequence_number`` for the row, that value is returned;
+otherwise the value is the data file's data sequence number.  Engines that
+preserve row lineage store each row's original value when compaction rewrites
+the row, so compaction does not change the column.
 
 .. code-block:: sql
 

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergColumnHandle.java

@@ -272,6 +272,11 @@ public boolean isLastUpdatedSequenceNumberColumn()
         return getColumnIdentity().getId() == LAST_UPDATED_SEQUENCE_NUMBER.fieldId();
     }
 
+    public boolean isRowLineageColumn()
+    {
+        return isRowIdColumn() || isLastUpdatedSequenceNumberColumn();
+    }
+
     public static IcebergColumnHandle primitiveIcebergColumnHandle(int id, String name, Type type, Optional<String> comment)
     {
         return new IcebergColumnHandle(primitiveColumnIdentity(id, name), type, comment, REGULAR);

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergPageSourceProvider.java

@@ -476,6 +476,10 @@ private static TupleDomain<ColumnDescriptor> getParquetTupleDomain(Map<List<Stri
 
         ImmutableMap.Builder<ColumnDescriptor, Domain> predicate = ImmutableMap.builder();
         effectivePredicate.getDomains().get().forEach((columnHandle, domain) -> {
+            // Stored row lineage values exclude inherited ones, so their statistics cannot prune.
+            if (columnHandle.isRowLineageColumn()) {
+                return;
+            }
             String baseType = columnHandle.getType().getTypeSignature().getBase();
             // skip looking up predicates for complex types as Parquet only stores stats for primitives
             if (!baseType.equals(StandardTypes.MAP) && !baseType.equals(StandardTypes.ARRAY) && !baseType.equals(StandardTypes.ROW)) {
@@ -621,10 +625,11 @@ private static ConnectorPageSourceWithRowPositions createBatchOrcPageSource(
                 }
             }
 
-            // Skip the time type columns in predicate, converted on page source level
+            // Skip the time type columns in predicate, converted on page source level, and the row
+            // lineage columns, whose stored values exclude inherited ones
             ImmutableMap.Builder<IcebergColumnHandle, Domain> predicateExcludeTimeType = ImmutableMap.builder();
             effectivePredicate.getDomains().get().forEach((columnHandle, domain) -> {
-                if (!(columnHandle.getType() instanceof TimeType)) {
+                if (!(columnHandle.getType() instanceof TimeType) && !columnHandle.isRowLineageColumn()) {
                     predicateExcludeTimeType.put(columnHandle, domain);
                 }
             });

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergSplitSource.java

@@ -21,11 +21,13 @@
 import com.facebook.presto.spi.SplitWeight;
 import com.facebook.presto.spi.connector.ConnectorPartitionHandle;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
+import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableList;
 import com.google.common.io.Closer;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.PartitionSpecParser;
+import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.TableScan;
 import org.apache.iceberg.expressions.InclusiveMetricsEvaluator;
@@ -38,13 +40,16 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
 
 import static com.facebook.presto.hive.HiveCommonSessionProperties.getAffinitySchedulingFileSectionSize;
 import static com.facebook.presto.hive.HiveCommonSessionProperties.getNodeSelectionStrategy;
 import static com.facebook.presto.iceberg.FileFormat.fromIcebergFileFormat;
 import static com.facebook.presto.iceberg.IcebergSessionProperties.getMinimumAssignedSplitWeight;
 import static com.facebook.presto.iceberg.IcebergUtil.buildLastUpdatedSequenceNumberEvaluator;
+import static com.facebook.presto.iceberg.IcebergUtil.getAppendSequenceNumbers;
 import static com.facebook.presto.iceberg.IcebergUtil.getDataSequenceNumber;
 import static com.facebook.presto.iceberg.IcebergUtil.getFirstRowId;
 import static com.facebook.presto.iceberg.IcebergUtil.getPartitionKeys;
@@ -70,6 +75,7 @@ public class IcebergSplitSource
 
     private final TupleDomain<IcebergColumnHandle> metadataColumnConstraints;
     private final InclusiveMetricsEvaluator lineageEvaluator;
+    private final Supplier<Set<Long>> appendSequenceNumbers;
     // Preferred Presto FileFormat the table is configured to write
     // (`write.format.default`), or null when the property is absent /
     // unrecognized. Used to disambiguate the iceberg-api wire format on
@@ -84,28 +90,32 @@ public IcebergSplitSource(
             TableScan tableScan,
             TupleDomain<IcebergColumnHandle> metadataColumnConstraints)
     {
-        this(session, getTargetSplitSize(session, tableScan).toBytes(), tableScan.planFiles(), metadataColumnConstraints, parseTableWriteFormat(tableScan));
+        this(session, tableScan.table(), getTargetSplitSize(session, tableScan).toBytes(), tableScan.planFiles(), metadataColumnConstraints, parseTableWriteFormat(tableScan));
     }
 
     public IcebergSplitSource(
             ConnectorSession session,
+            Table table,
             long targetSplitSize,
             CloseableIterable<FileScanTask> fileScanTasks,
             TupleDomain<IcebergColumnHandle> metadataColumnConstraints)
     {
-        this(session, targetSplitSize, fileScanTasks, metadataColumnConstraints, null);
+        this(session, table, targetSplitSize, fileScanTasks, metadataColumnConstraints, null);
     }
 
     private IcebergSplitSource(
             ConnectorSession session,
+            Table table,
             long targetSplitSize,
             CloseableIterable<FileScanTask> fileScanTasks,
             TupleDomain<IcebergColumnHandle> metadataColumnConstraints,
             FileFormat tableWriteFormat)
     {
         requireNonNull(session, "session is null");
+        requireNonNull(table, "table is null");
         this.metadataColumnConstraints = requireNonNull(metadataColumnConstraints, "metadataColumnConstraints is null");
         this.lineageEvaluator = buildLastUpdatedSequenceNumberEvaluator(metadataColumnConstraints);
+        this.appendSequenceNumbers = Suppliers.memoize(() -> getAppendSequenceNumbers(table));
         this.targetSplitSize = targetSplitSize;
         this.minimumAssignedSplitWeight = getMinimumAssignedSplitWeight(session);
         this.nodeSelectionStrategy = getNodeSelectionStrategy(session);
@@ -152,7 +162,8 @@ public CompletableFuture<ConnectorSplitBatch> getNextBatch(ConnectorPartitionHan
                     icebergSplit.getPath(),
                     icebergSplit.getDataSequenceNumber(),
                     task.file(),
-                    lineageEvaluator)) {
+                    lineageEvaluator,
+                    appendSequenceNumbers)) {
                 splits.add(icebergSplit);
             }
         }

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergUtil.java

@@ -20,7 +20,9 @@
 import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.predicate.Domain;
 import com.facebook.presto.common.predicate.NullableValue;
+import com.facebook.presto.common.predicate.Range;
 import com.facebook.presto.common.predicate.TupleDomain;
+import com.facebook.presto.common.predicate.ValueSet;
 import com.facebook.presto.common.type.DecimalType;
 import com.facebook.presto.common.type.Decimals;
 import com.facebook.presto.common.type.RealType;
@@ -69,6 +71,7 @@
 import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.ContentScanTask;
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataOperations;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.HistoryEntry;
@@ -797,12 +800,13 @@ protected static NullableValue parsePartitionValue(
     }
 
     // Row-lineage columns are stripped because Iceberg's Binder rejects field ids absent from
-    // the table schema; they are evaluated at the split level instead.
+    // the table schema. Splits are pruned on _last_updated_sequence_number only (see
+    // metadataColumnsMatchPredicates); _row_id is filtered per row.
     public static <U> TupleDomain<IcebergColumnHandle> getNonMetadataColumnConstraints(TupleDomain<U> allConstraints)
     {
         return allConstraints.transform(c -> {
             IcebergColumnHandle handle = (IcebergColumnHandle) c;
-            if (isMetadataColumnId(handle.getId()) || handle.isRowIdColumn() || handle.isLastUpdatedSequenceNumberColumn()) {
+            if (isMetadataColumnId(handle.getId()) || handle.isRowLineageColumn()) {
                 return null;
             }
             return handle;
@@ -825,7 +829,8 @@ public static boolean metadataColumnsMatchPredicates(
             String path,
             long dataSequenceNumber,
             ContentFile<?> file,
-            InclusiveMetricsEvaluator lineageEvaluator)
+            InclusiveMetricsEvaluator lineageEvaluator,
+            Supplier<Set<Long>> appendSequenceNumbers)
     {
         if (constraints.isAll()) {
             return true;
@@ -843,36 +848,64 @@ else if (handle == DATA_SEQUENCE_NUMBER_COLUMN_HANDLE) {
                     matches &= domain.includesNullableValue(dataSequenceNumber);
                 }
                 else if (handle.isLastUpdatedSequenceNumberColumn()) {
-                    matches &= lastUpdatedSequenceNumberMatches(domain, dataSequenceNumber, file, lineageEvaluator);
+                    matches &= lastUpdatedSequenceNumberMatches(domain, dataSequenceNumber, file, lineageEvaluator, appendSequenceNumbers);
                 }
             }
         }
 
         return matches;
     }
 
-    // The fallback branches handle cases where InclusiveMetricsEvaluator would over-include:
-    // V2/no-row-lineage files (column always null) and V3 pre-compaction (effective value =
-    // dataSequenceNumber per Iceberg's LastUpdatedSeqReader).
+    // A row's value is its stored value, or dataSequenceNumber when that is null, so a file matches if
+    // either can match. Stored values are known only through bounds. No value exceeds the file
+    // sequence number, since the spec has writers store only values copied from rows of earlier
+    // commits. A file with no metrics for the column is assumed to store none only if a retained
+    // APPEND snapshot added it. Iceberg reports APPEND only when a commit adds data files without
+    // removing data files or adding delete files, and writers are assumed, not guaranteed by the
+    // spec, to store lineage values only when they rewrite rows.
     private static boolean lastUpdatedSequenceNumberMatches(
             Domain domain,
             long dataSequenceNumber,
             ContentFile<?> file,
-            InclusiveMetricsEvaluator evaluator)
+            InclusiveMetricsEvaluator evaluator,
+            Supplier<Set<Long>> appendSequenceNumbers)
     {
-        int fieldId = LAST_UPDATED_SEQUENCE_NUMBER.fieldId();
-        Map<Integer, ByteBuffer> lowerBounds = file.lowerBounds();
-        Map<Integer, ByteBuffer> upperBounds = file.upperBounds();
-        if (lowerBounds != null && lowerBounds.containsKey(fieldId)
-                && upperBounds != null && upperBounds.containsKey(fieldId)) {
-            return evaluator.eval(file);
-        }
         if (file instanceof DataFile && ((DataFile) file).firstRowId() == null) {
             return domain.isNullAllowed();
         }
+        int fieldId = LAST_UPDATED_SEQUENCE_NUMBER.fieldId();
+        Long valueCount = getColumnMetric(file.valueCounts(), fieldId);
+        Long nullCount = getColumnMetric(file.nullValueCounts(), fieldId);
+        boolean mayHaveInheritedValues = nullCount == null || nullCount > 0;
+        if (getColumnMetric(file.lowerBounds(), fieldId) != null && getColumnMetric(file.upperBounds(), fieldId) != null) {
+            return evaluator.eval(file) || (mayHaveInheritedValues && domain.includesNullableValue(dataSequenceNumber));
+        }
+        Long fileSequenceNumber = file.fileSequenceNumber();
+        if (fileSequenceNumber != null && !domain.getValues().overlaps(ValueSet.ofRanges(Range.lessThanOrEqual(domain.getType(), fileSequenceNumber)))) {
+            return false;
+        }
+        if (valueCount != null && (nullCount == null || nullCount < valueCount)) {
+            return true;
+        }
+        if (valueCount == null && !appendSequenceNumbers.get().contains(fileSequenceNumber)) {
+            return true;
+        }
         return domain.includesNullableValue(dataSequenceNumber);
     }
 
+    public static Set<Long> getAppendSequenceNumbers(Table table)
+    {
+        return stream(table.snapshots())
+                .filter(snapshot -> DataOperations.APPEND.equals(snapshot.operation()))
+                .map(Snapshot::sequenceNumber)
+                .collect(toImmutableSet());
+    }
+
+    private static <T> T getColumnMetric(Map<Integer, T> metrics, int fieldId)
+    {
+        return metrics == null ? null : metrics.get(fieldId);
+    }
+
     public static InclusiveMetricsEvaluator buildLastUpdatedSequenceNumberEvaluator(TupleDomain<IcebergColumnHandle> metadataColumnConstraints)
     {
         if (metadataColumnConstraints.getDomains().isEmpty()) {

presto-iceberg/src/main/java/com/facebook/presto/iceberg/TableStatisticsMaker.java

@@ -311,9 +311,10 @@ private Partition getEqualityDeleteTableSummary(IcebergTableHandle tableHandle,
             List<Types.NestedField> nonPartitionPrimitiveColumns,
             List<PartitionField> partitionFields)
     {
+        // Iceberg's expression binder rejects predicates on metadata columns; strip them here.
         CloseableIterable<DeleteFile> deleteFiles = IcebergUtil.getDeleteFiles(icebergTable,
                 tableHandle.getIcebergTableName().getSnapshotId().get(),
-                intersection,
+                IcebergUtil.getNonMetadataColumnConstraints(intersection),
                 tableHandle.getPartitionSpecId(),
                 tableHandle.getEqualityFieldIds(),
                 session.getRuntimeStats());

presto-iceberg/src/main/java/com/facebook/presto/iceberg/procedure/splits/RewriteDataFilesIcebergSplitSource.java

@@ -40,7 +40,7 @@ public RewriteDataFilesIcebergSplitSource(
             TupleDomain<IcebergColumnHandle> metadataColumnConstraints,
             Map<String, String> options)
     {
-        super(session, getTargetSplitSize(session, tableScan).toBytes(), applyFilters(tableScan, options), metadataColumnConstraints);
+        super(session, tableScan.table(), getTargetSplitSize(session, tableScan).toBytes(), applyFilters(tableScan, options), metadataColumnConstraints);
     }
 
     private static CloseableIterable<FileScanTask> applyFilters(TableScan tableScan, Map<String, String> options)