apache / apache/gravitino

[Bug report] When using the Table Maintenance Service (Optimizer) and triggering rewrite submission with submit-strategy-jobs, the following error occurs: Cannot translate Spark expression:

Open
#10,587 1 comment 0 reactions 0 assignees View on GitHub
bug
Dominant language
Java
Stars
3.2k
Forks
935
Avg merge
1d 16h
Merged PRs (30d)
298

Description

### Version

main branch

### Describe what's wrong

When using the Table Maintenance Service (Optimizer) and triggering rewrite submission with submit-strategy-jobs, the following error occurs: Cannot translate Spark expression: (isnotnull(event_time#19) AND (day(cast(event_time#19 as date)) = 2005)) to data source filter.

### Error message and/or stacktrace

The log information in output.log is as follows:
Applied custom Spark configurations: {spark.master=local[2], spark.hadoop.fs.defaultFS=hdfs://196.105.0.49:9000}
Executing Iceberg rewrite_data_files procedure: CALL iceberg.system.rewrite_data_files(table => 'db.user_events', strategy => 'binpack', where => '(day(event_time) = 2026-03-18)')

The log information in error.log is as follows:
26/03/30 16:30:30 INFO BaseMetastoreCatalog: Table loaded by catalog: iceberg.db.user_events
Error executing rewrite data files job: Cannot translate Spark expression: (isnotnull(event_time#19) AND (day(cast(event_time#19 as date)) = 2005)) to data source filter
java.lang.IllegalArgumentException: Cannot translate Spark expression: (isnotnull(event_time#19) AND (day(cast(event_time#19 as date)) = 2005)) to data source filter
at org.apache.spark.sql.execution.datasources.SparkExpressionConverter$.convertToIcebergExpression(SparkExpressionConverter.scala:48)
at org.apache.spark.sql.execution.datasources.SparkExpressionConverter.convertToIcebergExpression(SparkExpressionConverter.scala)
at org.apache.iceberg.spark.procedures.BaseProcedure.filterExpression(BaseProcedure.java:171)
at org.apache.iceberg.spark.procedures.RewriteDataFilesProcedure.checkAndApplyFilter(RewriteDataFilesProcedure.java:131)
at org.apache.iceberg.spark.procedures.RewriteDataFilesProcedure.lambda$call$0(RewriteDataFilesProcedure.java:120)
at org.apache.iceberg.spark.procedures.BaseProcedure.execute(BaseProcedure.java:107)
at org.apache.iceberg.spark.procedures.BaseProcedure.modifyIcebergTable(BaseProcedure.java:88)
at org.apache.iceberg.spark.procedures.RewriteDataFilesProcedure.call(RewriteDataFilesProcedure.java:111)
at org.apache.spark.sql.execution.datasources.v2.CallExec.run(CallExec.scala:34)
at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result$lzycompute(V2CommandExec.scala:43)
at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result(V2CommandExec.scala:43)
at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.executeCollect(V2CommandExec.scala:49)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.$anonfun$applyOrElse$1(QueryExecution.scala:107)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:125)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:201)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:108)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:66)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:107)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:98)
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:461)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:76)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:461)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:32)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:32)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:32)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:437)
at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:98)
at org.apache.spark.sql.execution.QueryExecution.commandExecuted$lzycompute(QueryExecution.scala:85)
at org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:83)
at org.apache.spark.sql.Dataset.(Dataset.scala:220)
at org.apache.spark.sql.Dataset$.$anonfun$ofRows$2(Dataset.scala:100)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)
at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:97)
at org.apache.spark.sql.SparkSession.$anonfun$sql$4(SparkSession.scala:691)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)
at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:682)
at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:713)
at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:744)
at org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteDataFilesJob.main(IcebergRewriteDataFilesJob.java:173)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:75)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:52)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:1034)
at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:199)
at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:222)
at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:91)
at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1125)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1134)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

### How to reproduce

When executing the following command, an error occurs:
./bin/gravitino-optimizer.sh \
--type submit-strategy-jobs \
--identifiers iceberg.db.user_events \
--strategy-name iceberg_compaction_default \
--limit 10 \
--conf-path ./conf/gravitino-optimizer-submit.conf

The table structure is as follows:
CREATE TABLE IF NOT EXISTS user_events (
event_id BIGINT COMMENT 'Unique event ID',
user_id INT COMMENT 'User ID',
event_type STRING COMMENT 'Type of event',
event_time TIMESTAMP COMMENT 'Event timestamp',
event_data STRING COMMENT 'Detailed event data'
)
USING iceberg
PARTITIONED BY (days(event_time))
TBLPROPERTIES (
'format-version' = '2',
'write.upsert.enabled' = 'true'
)

### Additional context

_No response_

Contributor guide

Open the contributing guide

Research direction

Reproduce the optimizer command and start at IcebergRewriteDataFilesJob.java:173, then trace the predicate conversion through BaseProcedure.java:171 and SparkExpressionConverter.scala:48. Compare the generated day(event_time) filter with the failing stack trace; the work is done when the partitioned-table rewrite submission completes without the translation error.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, spark, sql
Domain
data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.