Hive: Filter pushdown by cmathiesen · Pull Request #1326 · apache/iceberg

cmathiesen · 2020-08-12T13:14:56Z

Hello! This is a follow-up PR to the Hive IF PR's that got merged recently that adds filter pushdown to the HiveIcebergInputFormat. I've added a filter factory to convert the Hive filter to an Iceberg Expression and then use the InputFormatConfig to set the filter expression for IcebergInputFormat to apply to the table scan.

cc: @rdblue @guilload @massdosage @pvary @rdsr @shardulm94

Thanks :D

RussellSpitzer · 2020-08-12T14:23:57Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+      case CONSTANT:
+        //We are unsure of how the CONSTANT case works, so using the approach of:
+        //https://github.com/apache/hive/blob/master/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/read/
+        // ParquetFilterPredicateConverter.java#L116


This comment seems to be at odds with the exception being thrown? It looks like in the hive code it just does nothing? Maybe I'm reading it wrong.

Ah yes, thanks for the spot - I changed the approach here during another review and didn't remove the comment. Will remove :)

RussellSpitzer · 2020-08-12T14:27:42Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+  private static Object leafToIcebergType(PredicateLeaf leaf) {
+    switch (leaf.getType()) {
+      case LONG:
+        return leaf.getLiteral() != null ? leaf.getLiteral() : leaf.getLiteralList();


I'm new to this code, so I wonder when reading this why we was to get the literal as a list if getLiteral is null? Does having getLiteral() returning null mean that there is a collection type?

Yeah, that's sort of whats going on. It looks like you'd only be using getLiteralList if the operator for the leaf was IN or BETWEEN, and then getLiteral for all other operator types. It would either be one or the other, so it seemed easiest to check for a null rather than calling getOperator and having cases to switch through all the different operators, if that makes sense

RussellSpitzer · 2020-08-12T14:30:09Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+      case TIMESTAMP:
+        if (leaf.getLiteral() != null) {
+          Timestamp timestamp = (Timestamp) leaf.getLiteral();
+          return timestamp.toInstant().getEpochSecond() * 1000000 + timestamp.getNanos() / 1000;


Not a big deal but I tend to make constants for numbers that can be misread like this, MILLION or MICROS_PER_SECOND

A very good point, I'll update these!

RussellSpitzer · 2020-08-12T14:41:54Z

mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergFilterFactory.java

+ * specific language governing permissions and limitations
+ * under the License.
+ */
+


I think we should probably have tests for all the filter literal types here, It seems like we are only checking Longs? Especially given the special code around other specific types.

I agree we should tests as many types as we can. We also don't need to test every predicate for every type. I think it's fine to test each predicate with a long and then to test each type with equals, for example.

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

rdblue · 2020-08-12T16:53:23Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+    }
+  }
+
+  private static Object leafToIcebergType(PredicateLeaf leaf) {


I think it would be better to split this into two methods: one for a single literal and one for a list of literals. Returning either one as Object doesn't allow us to make sure we're calling getLiteralList for the correct predicates.

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

rdsr · 2020-08-18T21:40:12Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+public class HiveIcebergFilterFactory {
+
+  private static final int MICROS_PER_SECOND = 1000000;
+  private static final int NANOSECS_PER_MICROSEC = 1000;


nit: to b consistent with MICROS_PER_SECOND maybe we can rename NANOSECS_PER_MICROSEC to NANOS_PER_MICROSEC

What about using TimeUnit for the conversion and get rid of those variables altogether?

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

rdsr · 2020-08-18T22:32:41Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+        //Hive converts a Date type to a Timestamp internally when retrieving literal
+        return ((Timestamp) leaf.getLiteral()).toLocalDateTime().toLocalDate().toEpochDay();
+      case DECIMAL:
+        return BigDecimal.valueOf(((HiveDecimalWritable) leaf.getLiteral()).doubleValue());


I'm unsure if this is correct. I think here, the scale of the BigDecimal will always be 0. Irrespective of the underlying data

Yes, this is not correct because it discards the scale and precision.

This should follow the examples from ORC, which also convert decimals: https://github.com/apache/iceberg/blob/master/data/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java#L163

rdsr · 2020-08-18T22:36:57Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+        dateValues.replaceAll(value -> ((Date) value).toLocalDate().toEpochDay());
+        return dateValues;
+      case DECIMAL:
+        List<Object> decimalValues = leaf.getLiteralList();


nit: I think it is clearer to not modify the returned list but to use standard idioms like leaf.getLiteralList().stream().map ... or Lists.transform(..)

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java

rdsr · 2020-08-18T22:40:17Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java

@@ -51,6 +58,17 @@ public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {

    forwardConfigSettings(job);


@cmathiesen the latest HiveIcebergInputFormat has changed substantially. Can you please rebase?

rebase

Shouldn't we be following the Golden Rule of Rebasing and not do this on public branches? It has the potential to cause all kinds of inconsistencies on other people's checkouts. Surely we should be doing merge? It all gets squash merged at the end so having a pristine history isn't worth the downsides of inconsistencies IHMO.

I think that either rebasing or merging master into a PR is okay.

As a reviewer, I don't really consider PR branches to be public because github handles force-pushes well. If I have a PR checked out, I also don't mind resetting to the PR's current state because I like keeping history clean.

That said, if you're sharing a PR branch between people that can be disruptive, so I think it is up to the author and collaborators whether to merge or to rebase to stay up to date with master.

Is that reasonable?

That sounds reasonable as long as the PR is rebased to master before it is committed. Having a linear history makes things much much easier to track where a change happened.

Agreed. We always merge by squashing the entire PR into a commit, so we do get a linear history in master.

api/src/main/java/org/apache/iceberg/ContentFile.java

cmathiesen · 2020-08-19T12:52:16Z

mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergFilterFactory.java

+    SearchArgument arg = builder.startAnd().equals("date", PredicateLeaf.Type.DATE,
+            Date.valueOf("2015-11-12")).end().build();
+
+    UnboundPredicate expected = Expressions.equal("date", LocalDate.of(2015,11,12));


@rdblue @rdsr I added tests for the Date and Timestamp types but when these are run I get errors like:

java.lang.IllegalArgumentException: Cannot create expression literal from java.time.LocalDate: 2015-11-12 at org.apache.iceberg.expressions.Literals.from(Literals.java:83) at org.apache.iceberg.expressions.UnboundPredicate.<init>(UnboundPredicate.java:39) at org.apache.iceberg.expressions.Expressions.equal(Expressions.java:159) at org.apache.iceberg.mr.hive.TestHiveIcebergFilterFactory.testDateType(TestHiveIcebergFilterFactory.java:211)

I noticed here in another test that Date's etc. are actually passed as Strings - is that the correct option to be using in this case?

Yes, it is a good idea to use a string instead of passing a LocalDate. The intent was to avoid tying the API to date/time representations from a specific library.

rdblue · 2020-08-19T17:52:56Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+      case IN:
+        return in(column, leafToLiteralList(leaf));
+      case BETWEEN:
+        List<Object> icebergLiterals = leafToLiteralList(leaf);


Do we need to validate that there are only two literals here, or is this reliable?

I believe that we can expect there to be 2, as we're using the BETWEEN operator and Hive wouldn't accept more than 2 arguments

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

cmathiesen · 2020-08-21T11:30:39Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+      case STRING:
+        return leaf.getLiteralList();
+      case DATE:
+        return leaf.getLiteralList().stream().map(value -> dateToString((Date) value))


I found a small quirk with the Hive Date type where if you call getLiteral you get a Timestamp back and if you call getLiteralList you get Date objects, which is why there are 2 separate methods for DATE

rdblue · 2020-08-21T16:32:47Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

        return leaf.getLiteralList().stream()
-                .map(value -> ((Timestamp) value).toInstant().getEpochSecond() * MICROS_PER_SECOND +
-                        ((Timestamp) value).getNanos() / NANOS_PER_MICROSEC).collect(Collectors.toList());
+                .map(value -> timestampToTimestampString((Timestamp) value))


This shouldn't convert to a string. Instead, it should convert the Timestamp value directly to microseconds from the unix epoch. String conversion in expressions is only for convenience in tests and for people using the API directly with generics. If an engine passes a predicate, we don't want to needlessly convert to string and back because it is much, much more likely to corrupt the value.

Ah sure, thank you for explaining that, I think I misunderstood what to do from the last comment - should hopefully be fixed now :)

rdblue · 2020-08-21T16:33:24Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+  }
+
+  private static BigDecimal hiveDecimalToBigDecimal(HiveDecimalWritable hiveDecimalWritable) {
+    return new BigDecimal(hiveDecimalWritable.toString()).setScale(hiveDecimalWritable.scale());


I don't think we want to convert to String here, either. Can you use the same logic from ORC?

pvary · 2020-08-23T09:30:44Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java

+      ExprNodeGenericFuncDesc exprNodeDesc = SerializationUtilities
+              .deserializeObject(hiveFilter, ExprNodeGenericFuncDesc.class);
+      SearchArgument sarg = ConvertAstToSearchArg.create(job, exprNodeDesc);
+      Expression filter = HiveIcebergFilterFactory.generateFilterExpression(sarg);


HiveIcebergFilterFactory.generateFilterExpression might throw UnsupportedOperationException.
Maybe it would be good to catch the exception and continue without filters in case if there is an error.
Hive runs the filters later anyway, so it will not cause issue.

rdblue · 2020-08-24T18:54:49Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+
+  private static long timestampToUnixEpoch(Timestamp timestamp) {
+    return timestamp.toInstant().getEpochSecond() * TimeUnit.SECONDS.toMicros(1) +
+            timestamp.getNanos() / TimeUnit.MICROSECONDS.toNanos(1);


This seems odd to me. Why not call TimeUnit.SECONDS.toMicros(timestamp.toInstant().getEpochSecond())? Using the toMicros function to get the conversion factor, but not actually using it for conversion is strange.

+1, similarly for timestamp.getNanos() / TimeUnit.MICROSECONDS.toNanos(1) -TimeUnit.NANOSECONDS.toMicros(timestamp.getNanos())

Ah yep, probably should have spotted that one 😅

rdblue · 2020-08-24T18:55:18Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+  }
+
+  private static String timestampToDateString(Timestamp timestamp) {
+    return timestamp.toLocalDateTime().toLocalDate().toString();


Dates need to be converted directly to a value and not a string also. You can use DateTimeUtil if you need.

rdblue · 2020-08-24T18:56:56Z

mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergFilterFactory.java

+    SearchArgument arg = builder.startAnd().equals("date", PredicateLeaf.Type.DATE,
+            Date.valueOf("2015-11-12")).end().build();
+
+    UnboundPredicate expected = Expressions.equal("date", "2015-11-12");


I think this expression should use an integer value instead of a String.

rdsr · 2020-08-25T08:24:15Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+
+  private static long timestampToUnixEpoch(Timestamp timestamp) {
+    return timestamp.toInstant().getEpochSecond() * TimeUnit.SECONDS.toMicros(1) +
+            timestamp.getNanos() / TimeUnit.MICROSECONDS.toNanos(1);


+1, similarly for timestamp.getNanos() / TimeUnit.MICROSECONDS.toNanos(1) -TimeUnit.NANOSECONDS.toMicros(timestamp.getNanos())

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java

rdsr · 2020-08-25T08:34:43Z

mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergFilterFactory.java

+  }
+
+  @Test
+  public void testAndOperand() {


Do we need a test using HiveRunner ? Since Hive stores the table's schema in lowercase I think we might have to support a case insensitive match on the iceberg side.
cc @pvary, @guilload

We definitely need to test for the lowercase column names, since Hive uses that. It might worth to do it for the InputFormat checks as well. On the other hand I am not sure if HiveRunner helps here or not.

I've been working on a HiveRunner test to see what happens in this case:

I've got an Iceberg table with a schema like:

private static final Schema STOCK_LIST_SCHEMA = new Schema( required(1, "ITEM_ID", Types.LongType.get()), required(2, "ITEM_COUNT", Types.LongType.get()) );

If I run a regular query either like SELECT ITEM_ID from default.stock_table or SELECT item_id from default.stock_table then this error occurs:

Caused by: java.lang.RuntimeException: cannot find field item_id from [org.apache.iceberg.mr.hive.serde.objectinspector.IcebergRecordObjectInspector$IcebergRecordStructField@c0fc462a, org.apache.iceberg.mr.hive.serde.objectinspector.IcebergRecordObjectInspector$IcebergRecordStructField@275a564e] at org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils.getStandardStructFieldRef(ObjectInspectorUtils.java:523) at org.apache.iceberg.mr.hive.serde.objectinspector.IcebergRecordObjectInspector.getStructFieldRef(IcebergRecordObjectInspector.java:68) at org.apache.hadoop.hive.ql.exec.ExprNodeColumnEvaluator.initialize(ExprNodeColumnEvaluator.java:56) at org.apache.hadoop.hive.ql.exec.Operator.initEvaluators(Operator.java:1033) at org.apache.hadoop.hive.ql.exec.Operator.initEvaluatorsAndReturnStruct(Operator.java:1059) at org.apache.hadoop.hive.ql.exec.SelectOperator.initializeOp(SelectOperator.java:75) at org.apache.hadoop.hive.ql.exec.Operator.initialize(Operator.java:366) at org.apache.hadoop.hive.ql.exec.Operator.initialize(Operator.java:556) at org.apache.hadoop.hive.ql.exec.Operator.initializeChildren(Operator.java:508) at org.apache.hadoop.hive.ql.exec.Operator.initialize(Operator.java:376) at org.apache.hadoop.hive.ql.exec.FetchTask.initialize(FetchTask.java:88) ... 29 more

which looks like the case sensitivity issues @rdsr mentioned.

I haven't pushed this test yet but I can do so if others want to reproduce the issue (I've just added a test to HiveIcebergStorageHandlerBaseTest).

Where would be the best place to put in a fix for this? This also doesn't rely on predicate pushdown so it could be done in another PR if needed

That doesn't look like a pushdown problem, so I'd open a separate PR to fix it and add the tests.

Sure, sounds good! I think I've addressed all the other comments on this PR so do you have time for another review? @rdblue @rdsr

pvary · 2020-08-27T09:20:55Z

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java

+  }
+
+  public static Expression generateFilterExpression(SearchArgument sarg) {
+    return translate(sarg.getExpression(), sarg.getLeaves());


Maybe logging would be nice here minimally on DEBUG level, but maybe on INFO level, like:

LOG.info("Translated sarg=[{}] to expression=[{}]", sarg, expression);

Not sure about the toString implementations, but the general idea would be to see what went in and what came out.
Also we can add this later, just noting here so we do not forget :D

Pushed filters are logged in the scan, so the translated expression is already logged. I assume that Hive also logs the filters that it is pushing, so I don't think this is necessary.

rdblue · 2020-08-30T01:10:55Z

@cmathiesen, I had a closer look at the date/time and decimal conversion and found that there were a few bugs. I opened ExpediaGroup#16 with the fixes for those problems. Could you review that and merge?

One major take-away was that it is not safe to call Timestamp.toLocalDate for conversion because that conversion is in local time, not UTC. FYI @massdosage, @rdsr, and @guilload.

This also hits HIVE-19726, which erases milliseconds. It was fixed in Hive 2.4.0, but I've added a work-around in the PR.

Fix date and timestamp conversion

rdblue · 2020-09-01T01:27:55Z

Thanks @cmathiesen! Merged. And thanks to all the reviewers!

cmathiesen added 3 commits July 27, 2020 17:24

Add filter factory

e2007c1

Fix type conversions

f97e408

Add tests

03729f4

RussellSpitzer reviewed Aug 12, 2020

View reviewed changes

rdblue reviewed Aug 12, 2020

View reviewed changes

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java Outdated Show resolved Hide resolved

rdblue reviewed Aug 12, 2020

View reviewed changes

Add more tests and change literalList method

b34300a

rdsr reviewed Aug 18, 2020

View reviewed changes

cmathiesen added 7 commits August 19, 2020 10:54

Add filter factory

45197f0

Fix type conversions

a1f67ac

Add tests

6c90876

Add more tests and change literalList method

e53592d

Use stream.map() and nit cleanup

9f3d598

Fix merge conflicts

cf7551d

PR review changes

4b8ca3e

probot-autolabeler bot added API core labels Aug 19, 2020

massdosage reviewed Aug 19, 2020

View reviewed changes

api/src/main/java/org/apache/iceberg/ContentFile.java Show resolved Hide resolved

massdosage force-pushed the hive-filter-pushdown branch from 0b9c046 to 4b8ca3e Compare August 19, 2020 11:48

probot-autolabeler bot added the MR label Aug 19, 2020

Merge branch 'master' into hive-filter-pushdown

d64a1dc

cmathiesen commented Aug 19, 2020

View reviewed changes

rdblue reviewed Aug 19, 2020

View reviewed changes

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergFilterFactory.java Show resolved Hide resolved

Conversion methods and fix date/timestamp tests

81fb5cb

cmathiesen commented Aug 21, 2020

View reviewed changes

rdblue reviewed Aug 21, 2020

View reviewed changes

pvary reviewed Aug 23, 2020

View reviewed changes

cmathiesen added 2 commits August 24, 2020 11:33

Correct timestamp and decimal conversion

4396672

Change log message to warn

848a3ab

rdblue reviewed Aug 24, 2020

View reviewed changes

rdsr reviewed Aug 25, 2020

View reviewed changes

cmathiesen added 3 commits August 25, 2020 09:43

Change date conversions

db95c5c

timeunit nanos to micros

b675224

fix faulty timestamp test

e97a461

pvary reviewed Aug 27, 2020

View reviewed changes

Fix date and timestamp conversion.

cfadba5

cmathiesen added 2 commits August 31, 2020 15:49

Merge pull request #16 from rdblue/pr-1326-hive-ppd

614781e

Fix date and timestamp conversion

Remove unused import

ef5b050

rdblue merged commit c801a2c into apache:master Sep 1, 2020

qphien mentioned this pull request Nov 2, 2020

Failed to select decimal value with hive when scale in filter is different with hive DDL schema #1699

Closed

rdblue added this to the Java 0.10.0 Release milestone Nov 16, 2020

		@@ -51,6 +58,17 @@ public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {

		forwardConfigSettings(job);

Conversation

cmathiesen commented Aug 12, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RussellSpitzer Aug 12, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RussellSpitzer Aug 12, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

rdsr Aug 18, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RussellSpitzer Aug 12, 2020 •

edited

Loading

RussellSpitzer Aug 12, 2020 •

edited

Loading

rdsr Aug 18, 2020 •

edited

Loading

rdsr Aug 25, 2020 •

edited

Loading