Hi @Steffen ,

To_timestamp() is a function. Let me show you on example. I recreated your data as you can see in below screen:

Now, for the sake of example I didn't apply any functions to attibutes and the filter pushdown works as expected:

df = spark.sql("""
    SELECT
        ts,
        id,
        AVG(value) AS avg
    FROM raw_measurements  
    GROUP BY
    id,
    ts
""")

df.createOrReplaceTempView("test_view")

query = f"""
SELECT *
FROM test_view
WHERE id = 1 AND ts BETWEEN 1751328004 AND 1751328104
"""
#display(spark.sql(query))
# Show logical/physical plan to inspect pushdown
spark.sql(query).explain(True)

Now, let's try to apply to_timestamp function to ts attribute:

df = spark.sql("""
    SELECT    
        id,    
        to_timestamp(from_unixtime(FLOOR((ts - 1) / 60) * 60 + 60)) as ts,
        AVG(value) AS avg
    FROM raw_measurements  
    GROUP BY
    id,
    ts
""")

df.createOrReplaceTempView("test_view")

query = f"""
SELECT *
FROM test_view
WHERE id = 1 AND ts BETWEEN  '2025-07-01T00:54:00.000+00:00' AND '2025-07-02T00:54:00.000+00:00'
"""
#display(spark.sql(query))
# Show logical/physical plan to inspect pushdown
spark.sql(query).explain(True)

As you can see, applying simple function to an attribute can prevent Spark SQL optimizer to kick in:

Your second example is a bit different though:

SELECT
    id,
    ts,
    AVG(value) OVER (ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS avg_value,
FROM
    measurements

Here I believe you encounter similar case to one described at below link:

[SPARK-23985] predicate push down doesn't work with simple compound partition spec - ASF JIRA

In short, filters are getting pushed only if they appear in the partitionSpec of window function. So, when you're using it like this:

df = spark.sql("""
    SELECT
        id,
        ts,
        AVG(value) OVER (ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS avg_value
    FROM
        raw_measurements
""")

df.createOrReplaceTempView("test_view")


query = f"""
SELECT *
FROM test_view
WHERE id = 1 AND ts BETWEEN 1751328004 AND 1751328104
"""
#display(spark.sql(query))
# Show logical/physical plan to inspect pushdown
spark.sql(query).explain(True)

Then pushed filters don't work:

But when you add ts attribute to partition by clause then optimizer will do its job:

And finally, what I recommended to try was to calculate at upstream table following attribute:

to_timestamp(from_unixtime(FLOOR((ts - 1) / 60) * 60 + 60)) as ts_aligned

And then create a view:

df = spark.sql("""
    SELECT
        ts_aligned,
        id,
        AVG(value) AS avg
    FROM
        raw_measurements
    GROUP BY
    id,
    ts_aligned
""")

df.createOrReplaceTempView("test_view")

Now, optimizer is again able to push filter to source: