[adapters] Use table-level skip_unused_columns property.

ryzhyk · ryzhyk · commit 576dd129214e · 2026-01-19T20:48:30.000Z
The Delta connector `skip_unused_columns` attribute introduces unhealthy
coupling between connector and table definitions, where the table definition
can implicitly change when a connector is added, removed, or modified.

We make skip_unused_columns a table-level property instead. It advises
connectors that they don't have to ingest unused columns if they support this
optimization.

We partially keep backward compatibility by computing `skip_unused_columns` as
a logical OR of per-connector and per-table attributes:

* Existing tables that don't have Delta connectors with `skip_unused_columns`
  set to true are not affected.
* Existing tables that have Delta connectors with `skip_unudes_columns` will
  behave as if the table definition has changed when the program is recompiled
  with the new runtime version. These tables will be backfilled from scratch.

Signed-off-by: Leonid Ryzhyk &lt;ryzhyk@gmail.com&gt;
diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs
@@ -519,6 +519,13 @@ impl DeltaTableInputEndpointInner {
         }
     }
 
+    fn skip_unused_columns(&self) -> bool {
+        // Old-style: property in the connector configuration.
+        // New-style: property in the table definition.
+        self.config.skip_unused_columns
+            || self.schema.get_property("skip_unused_columns") == Some("true")
+    }
+
     fn allocate_follow_transaction_label(&self) -> Option<Option<String>> {
         if self.config.transaction_mode == DeltaTableTransactionMode::Always {
             Some(Some(format!(
@@ -604,7 +611,7 @@ impl DeltaTableInputEndpointInner {
     /// table declaration.
     fn used_columns(&self, table: &DeltaTable) -> Vec<String> {
         // Column names in the SQL schema.
-        let sql_columns = if self.config.skip_unused_columns {
+        let sql_columns = if self.skip_unused_columns() {
             self.schema
                 .fields
                 .iter()
diff --git a/crates/feldera-types/src/program_schema.rs b/crates/feldera-types/src/program_schema.rs
@@ -238,6 +238,10 @@ impl Relation {
     pub fn has_lateness(&self) -> bool {
         self.fields.iter().any(|f| f.lateness.is_some())
     }
+
+    pub fn get_property(&self, name: &str) -> Option<&str> {
+        self.properties.get(name).map(|p| p.value.as_str())
+    }
 }
 
 /// A SQL field.
diff --git a/docs.feldera.com/docs/changelog.md b/docs.feldera.com/docs/changelog.md
@@ -15,9 +15,9 @@ import TabItem from '@theme/TabItem';
 
         ## Unreleased
 
-        Prometheus metrics output now also contains pipeline names with a
-        "pipeline_name" label, in addition to the exist "pipeline" label,
-        which still contains the pipeline UUID.
+        The Delta Lake connector's `skip_unused_columns` property has been deprecated. Use
+        table-level [`skip_unused_colums`](https://docs.feldera.com/sql/grammar#skip-unused-columns)
+        instead.
 
         ## 0.201.0
 
@@ -36,7 +36,13 @@ import TabItem from '@theme/TabItem';
         for the monitoring, is configured with an additional RBAC permission needed for this
         feature (see `kubernetes-runner-rbac.yaml` for changes).
 
-        ## 0.186.00
+        ## 0.188.0
+
+        Prometheus metrics output now also contains pipeline names with a
+        "pipeline_name" label, in addition to the exist "pipeline" label,
+        which still contains the pipeline UUID.
+
+        ## 0.186.0
 
         The Kafka input connector will now start reading partitions added
         to a topic upon resuming from a checkpoint.  Previously, the
diff --git a/docs.feldera.com/docs/connectors/sources/delta.md b/docs.feldera.com/docs/connectors/sources/delta.md
@@ -36,7 +36,7 @@ exactly once fault tolerance.
 | `cdc_delete_filer`          | string |            | <p>A predicate that determines whether the record represents a deletion.</p><p>This setting is only valid in the `cdc` mode. It specifies a predicate applied to each row in the Delta table to determine whether the row represents a deletion event. Its value must be a valid Boolean SQL expression that can be used in a query of the form `SELECT * from <table> WHERE <cdc_delete_filter>`.</p>|
 | `cdc_order_by`              | string |            | <p>An expression that determines the ordering of updates in the Delta table.</p><p>This setting is only valid in the `cdc` mode. It specifies a predicate applied to each row in the Delta table to determine the order in which updates in the table should be applied. Its value must be a valid SQL expression that can be used in a query of the form `SELECT * from <table> ORDER BY <cdc_order_by>`.</p>|
 | `num_parsers`               | string |            | The number of parallel parsing tasks the connector uses to process data read from the table. Increasing this value can enhance performance by allowing more concurrent processing. Recommended range: 1–10. The default is 4.|
-| `skip_unused_columns`       | bool   | false      | <b>Deprecated</b>, use the per-table property. <p>Don't read unused columns from the Delta table.  When set to `true`, this option instructs the connector to avoid reading columns from the Delta table that are not used in any view definitions. To be skipped, the columns must be either nullable or have default values. This can improve ingestion performance, especially for wide tables.</p><p>Note: The simplest way to exclude unused columns is to omit them from the Feldera SQL table declaration. The connector never reads columns that aren't declared in the SQL schema. Additionally, the SQL compiler emits warnings for declared but unused columns—use these as a guide to optimize your schema.</p>|
+| `skip_unused_columns` (<b>DEPRECATED</b>) | bool   | false | <p>This property is deprecated. Use the [table-level `skip_unused_columns` property](/sql/grammar#skip-unused-columns) instead.</p><p>Don't read unused columns from the Delta table.  When set to `true`, this option instructs the connector to avoid reading columns from the Delta table that are not used in any view definitions. To be skipped, the columns must be either nullable or have default values. This can improve ingestion performance, especially for wide tables.</p><p>Note: The simplest way to exclude unused columns is to omit them from the Feldera SQL table declaration. The connector never reads columns that aren't declared in the SQL schema. Additionally, the SQL compiler emits warnings for declared but unused columns—use these as a guide to optimize your schema.</p>|
 | `max_concurrent_readers`    | integer| 6          | <p>Maximum number of concurrent object store reads performed by all Delta Lake connectors.</p><p>This setting is used to limit the number of concurrent reads of the object store in a pipeline with a large number of Delta Lake connectors. When multiple connectors are simultaneously reading from the object store, this can lead to transport timeouts.</p><p>When enabled, this setting limits the number of concurrent reads across all connectors. This is a global setting that affects all Delta Lake connectors, and not just the connector where it is specified. It should therefore be used at most once in a pipeline.  If multiple connectors specify this setting, they must all use the same value.</p><p>The default value is 6.</p>|
 
 [*]: Required fields
diff --git a/docs.feldera.com/docs/sql/grammar.md b/docs.feldera.com/docs/sql/grammar.md
@@ -207,6 +207,7 @@ The property `expected_size` can be used to pass information to the
 SQL compiler about the expected size of a table in steady state
 operation.  The value of this property should be an integer value.
 
+<a id="skip-unused-columns"></a>
 #### Ignoring unused columns
 
 The `skip_unused_columns` is an optional Boolean property that can be

Original file line number	Diff line number	Diff line change
`@@ -238,6 +238,10 @@ impl Relation {`
`238`	`238`	`pub fn has_lateness(&self) -> bool {`
`239`	`239`	`self.fields.iter().any(\|f\| f.lateness.is_some())`
`240`	`240`	`}`
	`241`	`+`
	`242`	`+ pub fn get_property(&self, name: &str) -> Option<&str> {`
	`243`	`+ self.properties.get(name).map(\|p\| p.value.as_str())`
	`244`	`+ }`
`241`	`245`	`}`
`242`	`246`
`243`	`247`	`/// A SQL field.`