feat: DH-23045: Add pre-filter Iceberg pruning Expression

docs/groovy/how-to-guides/data-import-export/iceberg.md

@@ -367,6 +367,61 @@ snapshotInstructions = IcebergReadInstructions.builder()
     .build()
 ```
 
+#### Pruning expressions
+
+A pruning expression tells Iceberg to skip data files before Deephaven reads them. Set `pruningExpression` to an [`org.apache.iceberg.expressions.Expression`](https://iceberg.apache.org/javadoc/latest/org/apache/iceberg/expressions/Expression.html), and Deephaven discards every data file the expression proves cannot hold a matching row:
+
+```groovy docker-config=iceberg test-set=1 order=null
+import org.apache.iceberg.expressions.Expressions
+
+pruningInstructions = IcebergReadInstructions.builder()
+    .pruningExpression(Expressions.equal("store_and_fwd_flag", "Y"))
+    .build()
+
+forwardedTrips = icebergTaxis.table(pruningInstructions).where("StoreAndFwdFlag = `Y`")
+```
+
+Note the `where` in that example. Pruning is not filtering: Iceberg decides what to skip from partition values and per-file statistics, so a data file that survives is read in full, and the result is a superset of the rows that satisfy the expression. Apply an equivalent Deephaven filter whenever you need exactly the matching rows.
+
+Note also that the expression names `store_and_fwd_flag` while the filter names `StoreAndFwdFlag`. Field names in the expression are Iceberg schema names, so they are unaffected by column renames or by a resolver's table definition, and they are case-sensitive. An expression that names a field the Iceberg schema does not have is rejected when the table is read.
+
+> [!IMPORTANT]
+> Pruning on a non-partition field relies on the per-column value bounds that Iceberg records for each data file. Deephaven's Iceberg writer does not record those statistics, so an expression over a non-partition field prunes nothing on a table that Deephaven wrote. Pruning on a partition field applies in all cases, because Iceberg records partition values regardless of statistics.
+
+Iceberg accepts a Groovy numeric literal directly and widens it to the field's type. A temporal value has no such literal form and must be given as an explicit epoch offset through [`Literal`](https://iceberg.apache.org/javadoc/latest/org/apache/iceberg/expressions/Literal.html):
+
+```groovy order=null
+import io.deephaven.iceberg.util.IcebergReadInstructions
+import org.apache.iceberg.expressions.Expression
+import org.apache.iceberg.expressions.Expressions
+
+passengerInstructions = IcebergReadInstructions.builder()
+    .pruningExpression(Expressions.greaterThan("passenger_count", 2))
+    .build()
+
+// micros, millis, and nanos state the unit explicitly. Iceberg stores a timestamp column as
+// microseconds from the epoch unless the column is timestamp_ns.
+pickupInstructions = IcebergReadInstructions.builder()
+    .pruningExpression(
+        Expressions.predicate(
+            Expression.Operation.GT_EQ,
+            "tpep_pickup_datetime",
+            Expressions.micros(1704067200000000L)))
+    .build()
+```
+
+Use `Expressions.isNull` and `Expressions.notNull` to test for nulls; passing `null` as a value throws. Combine predicates with `Expressions.and`, `Expressions.or`, `Expressions.not`, and `Expressions.in`.
+
+To confirm that a pruning expression took effect, check the server log. Each time Deephaven discovers the data files of a pruned table, it logs a summary at INFO level:
+
+```
+IcebergFlatLayout[nyc.taxis]: pruning expression store_and_fwd_flag = (hash-…) skipped 3 manifest(s), could not be applied to 0 manifest(s), and accepted 2 data file(s)
+```
+
+Deephaven logs the expression through Iceberg's sanitized rendering, which keeps literal values out of the log: a string becomes an opaque `(hash-…)` fingerprint, a number becomes `(1-digit-int)`, and a timestamp becomes `(timestamp)`. Field names are printed as written.
+
+A non-zero count for "could not be applied to" means the expression referenced a field that did not yet exist when those manifests were written. Deephaven reads those manifests in full and logs a warning naming each one.
+
 ## Next steps
 
 This guide presented a basic example of interacting with an Iceberg catalog in Deephaven. These examples can be extended to include more complex queries, catalogs with multiple namespaces, snapshots, custom instructions, and more.

docs/groovy/reference/data-import-export/Iceberg/iceberg-read-instructions.md

@@ -15,6 +15,7 @@ import io.deephaven.iceberg.util.*
 instructions = IcebergReadInstructions.builder()
     .dataInstructions(s3Instructions)
     .ignoreResolvingErrors(ignoreResolvingErrors)
+    .pruningExpression(pruningExpression)
     .snapshot(snapshot)
     .snapshotId(snapshotId)
     .updateMode(updateMode)
@@ -27,6 +28,11 @@ The following parameters can be set using the builder:
 
 - `s3Instructions`: Instructions for accessing data in S3-compatible storage. Can be an arbitrary object, but is typically an instance of [`io.deephaven.extensions.s3.S3Instructions`](https://docs.deephaven.io/core/javadoc/io/deephaven/extensions/s3/S3Instructions.html).
 - `ignoreResolvingErrors`: Controls whether to ignore unexpected resolving errors by silently returning `null` data for columns that can't be resolved.
+- `pruningExpression`: An [`org.apache.iceberg.expressions.Expression`](https://iceberg.apache.org/javadoc/latest/org/apache/iceberg/expressions/Expression.html) that skips Iceberg data files that cannot contain matching rows. Field names resolve against the Iceberg schema rather than against Deephaven column names. The default is `Expressions.alwaysTrue`, which prunes nothing. This parameter prunes; it does not filter. Iceberg prunes using partition values and data file statistics, so a surviving data file is read in full and the result is a superset of the rows that satisfy the expression. To obtain exactly those rows, apply an equivalent Deephaven filter to the result.
+
+  > [!IMPORTANT]
+  > Pruning on a non-partition field relies on the per-column value bounds that Iceberg records for each data file. Deephaven's Iceberg writer does not record those statistics, so an expression over a non-partition field prunes nothing on a table that Deephaven wrote. Pruning on a partition field applies in all cases, because Iceberg records partition values regardless of statistics.
+
 - `snapshot`: The [`org.apache.iceberg.Snapshot`](https://iceberg.apache.org/javadoc/latest/org/apache/iceberg/Snapshot.html) to read. If not specified, the latest snapshot is used.
 - `snapshotId`: The ID of the snapshot to read. If not specified, the latest snapshot is used.
 - `updateMode`: The [`IcebergUpdateMode`](./iceberg-update-mode.md) to use when reading the table.
@@ -35,9 +41,11 @@ The following parameters can be set using the builder:
 
 - [`dataInstructions`](https://docs.deephaven.io/core/javadoc/io/deephaven/iceberg/util/IcebergReadInstructions.html#dataInstructions()): The data instructions to use for reading Iceberg data files.
 - [`ignoreResolvingErrors`](https://docs.deephaven.io/core/javadoc/io/deephaven/iceberg/util/IcebergReadInstructions.html#ignoreResolvingErrors()): Controls whether to ignore unexpected resolving errors by silently returning `null` data for columns that can't be resolved.
+- [`pruningExpression`](https://docs.deephaven.io/core/javadoc/io/deephaven/iceberg/util/IcebergReadInstructions.html#pruningExpression()): The Iceberg expression used to skip data files that cannot contain matching rows.
 - [`snapshot`](https://docs.deephaven.io/core/javadoc/io/deephaven/iceberg/util/IcebergReadInstructions.html#snapshot()): The snapshot to load for reading.
 - [`snapshotId`](https://docs.deephaven.io/core/javadoc/io/deephaven/iceberg/util/IcebergReadInstructions.html#snapshotId()): The snapshot ID to load for reading.
 - [`updateMode`](https://docs.deephaven.io/core/javadoc/io/deephaven/iceberg/util/IcebergReadInstructions.html#updateMode()): The [`IcebergUpdateMode`](./iceberg-update-mode.md) to use when reading Iceberg data files.
+- [`withPruningExpression`](https://docs.deephaven.io/core/javadoc/io/deephaven/iceberg/util/IcebergReadInstructions.html#withPruningExpression(org.apache.iceberg.expressions.Expression)): Return a copy of the instructions with the pruning expression replaced by the specified expression.
 - [`withSnapshot`](https://docs.deephaven.io/core/javadoc/io/deephaven/iceberg/util/IcebergReadInstructions.html#withSnapshot(org.apache.iceberg.Snapshot)): Return a copy of the instructions with the snapshot replaced by the specified snapshot.
 - [`withSnapshotId`](https://docs.deephaven.io/core/javadoc/io/deephaven/iceberg/util/IcebergReadInstructions.html#withSnapshotId(long)): Return a copy of the instructions with the snapshot ID replaced by the specified snapshot ID.
 
@@ -55,6 +63,39 @@ instructions = IcebergReadInstructions.builder()
     .build()
 ```
 
+The following example constructs an `IcebergReadInstructions` object that prunes data files whose `region` partition cannot contain the value `EMEA`. Because pruning is not filtering, the query applies an equivalent Deephaven filter to the result to obtain exactly the matching rows:
+
+```groovy
+import io.deephaven.iceberg.util.*
+import org.apache.iceberg.expressions.Expressions
+
+pruningInstructions = IcebergReadInstructions.builder()
+    .pruningExpression(Expressions.equal("region", "EMEA"))
+    .build()
+
+// emeaTable = tableAdapter.table(pruningInstructions).where("Region = `EMEA`")
+```
+
+Iceberg accepts a Groovy numeric literal directly and widens it to the field's type. A temporal value has no such literal form and must be given as an explicit epoch offset through [`Literal`](https://iceberg.apache.org/javadoc/latest/org/apache/iceberg/expressions/Literal.html). The following example prunes on an integer field and on a timestamp field:
+
+```groovy
+import io.deephaven.iceberg.util.*
+import org.apache.iceberg.expressions.Expression
+import org.apache.iceberg.expressions.Expressions
+
+yearInstructions = IcebergReadInstructions.builder()
+    .pruningExpression(Expressions.greaterThan("year", 2023))
+    .build()
+
+// micros, millis, and nanos state the unit explicitly. Iceberg stores a timestamp column as
+// microseconds from the epoch unless the column is timestamp_ns.
+timestampInstructions = IcebergReadInstructions.builder()
+    .pruningExpression(
+        Expressions.predicate(
+            Expression.Operation.GT_EQ, "pickup_time", Expressions.micros(1767225600000000L)))
+    .build()
+```
+
 ## Related documentation
 
 - [Deephaven and Iceberg](../../../how-to-guides/data-import-export/iceberg.md)

docs/groovy/snapshots/4148f4753ef862d27b298b04a85f226f.json

@@ -0,0 +1 @@
+{"file":"reference/data-import-export/Iceberg/iceberg-read-instructions.md","objects":{}}

docs/groovy/snapshots/725c3a9498a47c3d698a33c69dc07446.json

@@ -0,0 +1 @@
+{"file":"how-to-guides/data-import-export/iceberg.md","objects":{}}

docs/groovy/snapshots/ce873bfe60e7ffc02d7d2670a915f5cc.json

@@ -0,0 +1 @@
+{"file":"reference/data-import-export/Iceberg/iceberg-read-instructions.md","objects":{}}

docs/python/how-to-guides/data-import-export/iceberg.md

@@ -380,6 +380,75 @@ from deephaven.experimental import iceberg
 snapshot_instructions = iceberg.IcebergReadInstructions(snapshot_id=6738371110677246500)
 ```
 
+#### Pruning expressions
+
+A pruning expression tells Iceberg to skip data files before Deephaven reads them. Pass an [`org.apache.iceberg.expressions.Expression`](https://iceberg.apache.org/javadoc/latest/org/apache/iceberg/expressions/Expression.html) as `pruning_expression`, and Deephaven discards every data file the expression proves cannot hold a matching row:
+
+```python docker-config=iceberg test-set=1 order=null
+import jpy
+
+Expressions = jpy.get_type("org.apache.iceberg.expressions.Expressions")
+
+forwarded_trips = iceberg_taxis.table(
+    pruning_expression=Expressions.equal("store_and_fwd_flag", "Y")
+).where("StoreAndFwdFlag = `Y`")
+```
+
+Note the `where` in that example. Pruning is not filtering: Iceberg decides what to skip from partition values and per-file statistics, so a data file that survives is read in full, and the result is a superset of the rows that satisfy the expression. Apply an equivalent Deephaven filter whenever you need exactly the matching rows.
+
+Note also that the expression names `store_and_fwd_flag` while the filter names `StoreAndFwdFlag`. Field names in the expression are Iceberg schema names, so they are unaffected by `column_renames` or by a resolver's table definition, and they are case-sensitive. An expression that names a field the Iceberg schema does not have is rejected when the table is read.
+
+> [!IMPORTANT]
+> Pruning on a non-partition field relies on the per-column value bounds that Iceberg records for each data file. Deephaven's Iceberg writer does not record those statistics, so an expression over a non-partition field prunes nothing on a table that Deephaven wrote. Pruning on a partition field applies in all cases, because Iceberg records partition values regardless of statistics.
+
+Strings, floats, and booleans can be passed straight to `Expressions.equal` and its siblings, as above. Integers and timestamps need more care, because jpy converts a Python `int` to the narrowest Java box that holds it — usually a `Byte` or a `Short` — and Iceberg rejects both. Build those predicates from a typed [`Literal`](https://iceberg.apache.org/javadoc/latest/org/apache/iceberg/expressions/Literal.html) instead:
+
+```python order=null
+import jpy
+from deephaven.experimental import iceberg
+
+Expressions = jpy.get_type("org.apache.iceberg.expressions.Expressions")
+Literal = jpy.get_type("org.apache.iceberg.expressions.Literal")
+Operation = jpy.get_type("org.apache.iceberg.expressions.Expression$Operation")
+
+# Literal.of is overloaded on primitives, so the integer width survives the call into Java.
+passenger_instructions = iceberg.IcebergReadInstructions(
+    pruning_expression=Expressions.predicate(
+        Operation.GT, "passenger_count", Literal.of(2)
+    )
+)
+
+# micros, millis, and nanos state the unit explicitly. Iceberg stores a timestamp column as
+# microseconds from the epoch unless the column is timestamp_ns.
+pickup_instructions = iceberg.IcebergReadInstructions(
+    pruning_expression=Expressions.predicate(
+        Operation.GT_EQ, "tpep_pickup_datetime", Expressions.micros(1704067200000000)
+    )
+)
+```
+
+Use `Expressions.isNull` and `Expressions.notNull` to test for nulls; passing `None` as a value raises. The combinators `and`, `or`, `not`, and `in` are Python keywords, so reach them through `getattr`:
+
+```python order=null
+import jpy
+
+Expressions = jpy.get_type("org.apache.iceberg.expressions.Expressions")
+
+combined_expression = getattr(Expressions, "and")(
+    Expressions.equal("store_and_fwd_flag", "Y"), Expressions.notNull("trip_distance")
+)
+```
+
+To confirm that a pruning expression took effect, check the server log. Each time Deephaven discovers the data files of a pruned table, it logs a summary at INFO level:
+
+```
+IcebergFlatLayout[nyc.taxis]: pruning expression store_and_fwd_flag = (hash-…) skipped 3 manifest(s), could not be applied to 0 manifest(s), and accepted 2 data file(s)
+```
+
+Deephaven logs the expression through Iceberg's sanitized rendering, which keeps literal values out of the log: a string becomes an opaque `(hash-…)` fingerprint, a number becomes `(1-digit-int)`, and a timestamp becomes `(timestamp)`. Field names are printed as written.
+
+A non-zero count for "could not be applied to" means the expression referenced a field that did not yet exist when those manifests were written. Deephaven reads those manifests in full and logs a warning naming each one.
+
 ## Next steps
 
 This guide presented a basic example of interacting with an Iceberg catalog in Deephaven. These examples can be extended to include more complex queries, catalogs with multiple namespaces, snapshots, custom instructions, and more.

docs/python/reference/iceberg/iceberg-read-instructions.md

@@ -12,7 +12,9 @@ IcebergReadInstructions(
     data_instructions: S3Instructions = None,
     column_renames: Dict[str, str] = None,
     update_mode: IcebergUpdateMode = None,
-    snapshot_id: int = None
+    snapshot_id: int = None,
+    ignore_resolving_errors: bool = False,
+    pruning_expression: jpy.JType = None
 )
 ```
 
@@ -47,6 +49,21 @@ The update mode for the table. Options include:
 
 The snapshot ID to read. If not given, the most recent snapshot ID is used.
 
+</Param>
+<Param name="ignore_resolving_errors" type="bool" Optional>
+
+Controls whether to ignore unexpected resolving errors by silently returning `NULL` data for columns that cannot be resolved in the data files where they should be present. Such errors may indicate an incorrect resolver or name mapping, or an Iceberg metadata or data issue. The default is `False`.
+
+</Param>
+<Param name="pruning_expression" type="jpy.JType" Optional>
+
+An [`org.apache.iceberg.expressions.Expression`](https://iceberg.apache.org/javadoc/latest/org/apache/iceberg/expressions/Expression.html) that skips Iceberg data files that cannot contain matching rows. Field names resolve against the Iceberg schema rather than against Deephaven column names. The default is `Expressions.alwaysTrue`, which prunes nothing.
+
+This parameter prunes; it does not filter. Iceberg prunes using partition values and data file statistics, so a surviving data file is read in full and the result is a superset of the rows that satisfy the expression. To obtain exactly those rows, apply an equivalent Deephaven filter to the result.
+
+> [!IMPORTANT]
+> Pruning on a non-partition field relies on the per-column value bounds that Iceberg records for each data file. Deephaven's Iceberg writer does not record those statistics, so an expression over a non-partition field prunes nothing on a table that Deephaven wrote. Pruning on a partition field applies in all cases, because Iceberg records partition values regardless of statistics.
+
 </Param>
 </ParamTable>
 
@@ -126,6 +143,45 @@ iceberg_instructions = iceberg.IcebergReadInstructions(
 )
 ```
 
+The following example creates an `IcebergReadInstructions` object that prunes data files whose `region` partition cannot contain the value `EMEA`. Because pruning is not filtering, the query applies an equivalent Deephaven filter to the result to obtain exactly the matching rows:
+
+```python order=null
+import jpy
+from deephaven.experimental import iceberg
+
+Expressions = jpy.get_type("org.apache.iceberg.expressions.Expressions")
+
+pruning_instructions = iceberg.IcebergReadInstructions(
+    pruning_expression=Expressions.equal("region", "EMEA")
+)
+
+# emea_table = table_adapter.table(pruning_instructions).where("Region = `EMEA`")
+```
+
+Numeric and temporal literals require a typed [`Literal`](https://iceberg.apache.org/javadoc/latest/org/apache/iceberg/expressions/Literal.html), because jpy narrows a Python `int` to a Java `Byte` or `Short`, both of which Iceberg rejects. The following example prunes on an integer field and on a timestamp field:
+
+```python order=null
+import jpy
+from deephaven.experimental import iceberg
+
+Expressions = jpy.get_type("org.apache.iceberg.expressions.Expressions")
+Literal = jpy.get_type("org.apache.iceberg.expressions.Literal")
+Operation = jpy.get_type("org.apache.iceberg.expressions.Expression$Operation")
+
+# Literal.of accepts a primitive, so the integer width survives the call into Java.
+year_instructions = iceberg.IcebergReadInstructions(
+    pruning_expression=Expressions.predicate(Operation.GT, "year", Literal.of(2023))
+)
+
+# micros, millis, and nanos state the unit explicitly. Iceberg stores a timestamp column as
+# microseconds from the epoch unless the column is timestamp_ns.
+timestamp_instructions = iceberg.IcebergReadInstructions(
+    pruning_expression=Expressions.predicate(
+        Operation.GT_EQ, "pickup_time", Expressions.micros(1767225600000000)
+    )
+)
+```
+
 ## Related documentation
 
 - [`adapter`](./adapter.md)

docs/python/snapshots/4596bd7d90c69cf2e83f830933a44de1.json

@@ -0,0 +1 @@
+{"file":"how-to-guides/data-import-export/iceberg.md","objects":{}}

docs/python/snapshots/4a689cd0c9eeca869a546c82a3348c9f.json

@@ -0,0 +1 @@
+{"file":"reference/iceberg/iceberg-read-instructions.md","objects":{}}

docs/python/snapshots/b16c00f3c943c2405efd4074da27a7e2.json

@@ -0,0 +1 @@
+{"file":"reference/iceberg/iceberg-read-instructions.md","objects":{}}

docs/python/snapshots/cb353dd1b8857ba4992b49076edc8aeb.json

@@ -0,0 +1 @@
+{"file":"how-to-guides/data-import-export/iceberg.md","objects":{}}