You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "a2l007 (via GitHub)" <gi...@apache.org> on 2025/09/15 18:56:35 UTC
[PR] Add time window filter to Iceberg reader (druid)
a2l007 opened a new pull request, #18531:
URL: https://github.com/apache/druid/pull/18531
### Description
Introduces a time window filter to the Druid Iceberg reader which can be used to define lookback and lookahead durations while filtering on a partition time column.
With the default configuration, this filter will only read data files from the last 24 hours. Combining this filter with the scheduled batch supervisor, an ingestion pipeline can be setup in Druid that reads the latest iceberg data every day.
#### Release note
Adds the ability to filter iceberg data based on a configurable time window.
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [x] been self-reviewed.
- [x] added documentation for new or modified features or behaviors.
- [x] a release note entry in the PR description.
- [x] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [x] been tested in a test Druid cluster.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add time window filter to Iceberg reader (druid)
Posted by "a2l007 (via GitHub)" <gi...@apache.org>.
a2l007 commented on PR #18531:
URL: https://github.com/apache/druid/pull/18531#issuecomment-3300510107
> Do we rely on users to manually add those Iceberg filter as Druid filter in transformSpec? Out of scope of this PR, but I'm wondering if it's either possible to automatically convert Iceberg filter and add them to Druid filter in transformSpec for the users or warn/fail the job if the Iceberg filters result in residual files and there is no corresponding filters in transformSpec.
That's a good idea, I think if we're able to push down the transformSpec to the inputSource, it should be possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add time window filter to Iceberg reader (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on PR #18531:
URL: https://github.com/apache/druid/pull/18531#issuecomment-3299977331
> > How is this different than setting `intervals` in `granularitySpec`?
>
> @maytasm Thanks for taking a look! This filter is at the input source level which means it controls the data file paths pulled from the iceberg table for ingestion. Pruning the data files during the iceberg table scan phase helps reduce scan times and also reduces the number of files processed by the subsequent indexing phase.
I see. Not directly related to this PR (time window filter) but Iceberg filter in general, what if the filterColumn is not a partition column? This would result in files that may contain rows not matching the filter right? We would then need to filter it on Druid side again using filter in transformSpec?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add time window filter to Iceberg reader (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on PR #18531:
URL: https://github.com/apache/druid/pull/18531#issuecomment-3294500145
How is this different than setting `intervals` in `granularitySpec`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add time window filter to Iceberg reader (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #18531:
URL: https://github.com/apache/druid/pull/18531#discussion_r2349898935
##########
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/filter/IcebergTimeWindowFilter.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.iceberg.filter;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.types.Types;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Period;
+
+public class IcebergTimeWindowFilter implements IcebergFilter
+{
+ @JsonProperty
+ private final String filterColumn;
+
+ @JsonProperty
+ private final Duration lookbackDuration;
+
+ @JsonProperty
+ private final Duration lookaheadDuration;
+
+ @JsonProperty
+ private final DateTime baseTime;
+
+ @JsonCreator
+ public IcebergTimeWindowFilter(
+ @JsonProperty("filterColumn") String filterColumn,
+ @JsonProperty("lookbackDuration") Duration lookbackDuration,
+ @JsonProperty("lookaheadDuration") Duration lookaheadDuration,
+ @JsonProperty("baseTime") DateTime baseTime
+ )
+ {
+ Preconditions.checkNotNull(filterColumn, "You must specify a filter column on the timeWindow filter");
+ this.filterColumn = filterColumn;
+ this.lookbackDuration = Configs.valueOrDefault(lookbackDuration, new Period("P1D").toStandardDuration());
+ this.lookaheadDuration = Configs.valueOrDefault(lookaheadDuration, Duration.ZERO);
+ this.baseTime = Configs.valueOrDefault(baseTime, DateTimes.nowUtc());
+ }
+
+ @Override
+ public TableScan filter(TableScan tableScan)
+ {
+ return tableScan.filter(getFilterExpression());
+ }
+
+ @Override
+ public Expression getFilterExpression()
+ {
+ // Convert milliseconds to microseconds because Iceberg TimestampType uses microsecond precision
+ Long lookbackDurationinMicros = (baseTime.getMillis() - lookbackDuration.getMillis()) * 1000L;
+ Long lookforwardDurationinMicros = (baseTime.getMillis() + lookaheadDuration.getMillis()) * 1000L;
Review Comment:
## Boxed variable is never null
The variable 'lookforwardDurationinMicros' is only assigned values of primitive type and is never 'null', but it is declared with the boxed type 'Long'.
[Show more details](https://github.com/apache/druid/security/code-scanning/10332)
##########
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/filter/IcebergTimeWindowFilter.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.iceberg.filter;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.types.Types;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Period;
+
+public class IcebergTimeWindowFilter implements IcebergFilter
+{
+ @JsonProperty
+ private final String filterColumn;
+
+ @JsonProperty
+ private final Duration lookbackDuration;
+
+ @JsonProperty
+ private final Duration lookaheadDuration;
+
+ @JsonProperty
+ private final DateTime baseTime;
+
+ @JsonCreator
+ public IcebergTimeWindowFilter(
+ @JsonProperty("filterColumn") String filterColumn,
+ @JsonProperty("lookbackDuration") Duration lookbackDuration,
+ @JsonProperty("lookaheadDuration") Duration lookaheadDuration,
+ @JsonProperty("baseTime") DateTime baseTime
+ )
+ {
+ Preconditions.checkNotNull(filterColumn, "You must specify a filter column on the timeWindow filter");
+ this.filterColumn = filterColumn;
+ this.lookbackDuration = Configs.valueOrDefault(lookbackDuration, new Period("P1D").toStandardDuration());
+ this.lookaheadDuration = Configs.valueOrDefault(lookaheadDuration, Duration.ZERO);
+ this.baseTime = Configs.valueOrDefault(baseTime, DateTimes.nowUtc());
+ }
+
+ @Override
+ public TableScan filter(TableScan tableScan)
+ {
+ return tableScan.filter(getFilterExpression());
+ }
+
+ @Override
+ public Expression getFilterExpression()
+ {
+ // Convert milliseconds to microseconds because Iceberg TimestampType uses microsecond precision
+ Long lookbackDurationinMicros = (baseTime.getMillis() - lookbackDuration.getMillis()) * 1000L;
Review Comment:
## Boxed variable is never null
The variable 'lookbackDurationinMicros' is only assigned values of primitive type and is never 'null', but it is declared with the boxed type 'Long'.
[Show more details](https://github.com/apache/druid/security/code-scanning/10331)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add time window filter to Iceberg reader (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on PR #18531:
URL: https://github.com/apache/druid/pull/18531#issuecomment-3300282016
> > This would result in files that may contain rows not matching the filter right? We would then need to filter it on Druid side again using filter in transformSpec?
>
> That is correct, filtering on non partition columns may pull residual rows which will have to be filtered on the Druid side.
Do we rely on users to manually add those Iceberg filter as Druid filter in transformSpec? Out of scope of this PR, but I'm wondering if it's either possible to automatically convert Iceberg filter and add them to Druid filter in transformSpec for the users or warn/fail the job if the Iceberg filters result in residual files and there is no corresponding filters in transformSpec.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add time window filter to Iceberg reader (druid)
Posted by "a2l007 (via GitHub)" <gi...@apache.org>.
a2l007 merged PR #18531:
URL: https://github.com/apache/druid/pull/18531
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add time window filter to Iceberg reader (druid)
Posted by "a2l007 (via GitHub)" <gi...@apache.org>.
a2l007 commented on PR #18531:
URL: https://github.com/apache/druid/pull/18531#issuecomment-3299318217
> How is this different than setting `intervals` in `granularitySpec`?
@maytasm Thanks for taking a look! This filter is at the input source level which means it controls the data file paths pulled from the iceberg table for ingestion. Pruning the data files during the iceberg table scan phase helps reduce scan times and also reduces the number of files processed by the subsequent indexing phase.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add time window filter to Iceberg reader (druid)
Posted by "a2l007 (via GitHub)" <gi...@apache.org>.
a2l007 commented on PR #18531:
URL: https://github.com/apache/druid/pull/18531#issuecomment-3300249676
> This would result in files that may contain rows not matching the filter right? We would then need to filter it on Druid side again using filter in transformSpec?
That is correct, filtering on non partition columns may pull residual rows which will have to be filtered on the Druid side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org